package com.pig4cloud.pigx.strategy.executor.contract.task;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.okex.open.api.exception.APIException;
import com.pig4cloud.pigx.common.customize.enums.CommonDictEnum;
import com.pig4cloud.pigx.common.customize.exception.DuckweedException;
import com.pig4cloud.pigx.common.customize.exception.MqAckException;
import com.pig4cloud.pigx.common.customize.exception.MqAckLogException;
import com.pig4cloud.pigx.common.customize.exception.MqNotAckException;
import com.pig4cloud.pigx.common.customize.mysql.utils.MysqlUtil;
import com.pig4cloud.pigx.common.customize.mysql.utils.TransactionInterface;
import com.pig4cloud.pigx.common.customize.redis.constants.CommonRedisConstants;
import com.pig4cloud.pigx.common.customize.redis.constants.RedisKeyEnum;
import com.pig4cloud.pigx.common.customize.redis.utils.RedisUtils;
import com.pig4cloud.pigx.common.customize.security.utils.SecurityUtil;
import com.pig4cloud.pigx.common.customize.utils.BigDecimalUtils;
import com.pig4cloud.pigx.common.customize.utils.CommonUtil;
import com.pig4cloud.pigx.common.seata.utils.SeataUtils;
import com.pig4cloud.pigx.common.security.service.PigxUser;
import com.pig4cloud.pigx.strategy.bo.StrategyStopOrderContractCacheBO;
import com.pig4cloud.pigx.strategy.bo.StrategyTaskContractCacheBO;
import com.pig4cloud.pigx.strategy.bo.StrategyTaskContractCalculateProfitBO;
import com.pig4cloud.pigx.strategy.bo.okx.PlaceOrderBO;
import com.pig4cloud.pigx.strategy.bo.okx.SyncOrderBO;
import com.pig4cloud.pigx.strategy.constant.StrategyMqConstant;
import com.pig4cloud.pigx.strategy.constant.StrategyRedisLockConstant;
import com.pig4cloud.pigx.strategy.constant.StrategyScheduleConstant;
import com.pig4cloud.pigx.strategy.dto.StrategyContractCloseDTO;
import com.pig4cloud.pigx.strategy.dto.StrategyContractErrorDTO;
import com.pig4cloud.pigx.strategy.dto.StrategyContractErrorResolveDTO;
import com.pig4cloud.pigx.strategy.dto.StrategyContractErrorSettlementDTO;
import com.pig4cloud.pigx.strategy.dto.StrategyContractPauseDTO;
import com.pig4cloud.pigx.strategy.dto.StrategyContractRestoreDTO;
import com.pig4cloud.pigx.strategy.dto.StrategyOrderContractErrorDTO;
import com.pig4cloud.pigx.strategy.dto.StrategyPositionContractCloseDTO;
import com.pig4cloud.pigx.strategy.dto.StrategyPositionContractClosePartDTO;
import com.pig4cloud.pigx.strategy.dto.StrategyPositionContractOpenDTO;
import com.pig4cloud.pigx.strategy.dto.StrategyStopOrderContractAddDTO;
import com.pig4cloud.pigx.strategy.dto.StrategyStopOrderContractCancelBatchDTO;
import com.pig4cloud.pigx.strategy.dto.StrategySyncStopOrderMsgDTO;
import com.pig4cloud.pigx.strategy.entity.StrategyContract;
import com.pig4cloud.pigx.strategy.entity.StrategyGroupContract;
import com.pig4cloud.pigx.strategy.entity.StrategyOrderContract;
import com.pig4cloud.pigx.strategy.entity.StrategyPositionContract;
import com.pig4cloud.pigx.strategy.entity.StrategyStopOrderContract;
import com.pig4cloud.pigx.strategy.entity.StrategySwitchContract;
import com.pig4cloud.pigx.strategy.entity.StrategyTaskContract;
import com.pig4cloud.pigx.strategy.enums.StrategyContractEnum;
import com.pig4cloud.pigx.strategy.enums.StrategyOrderContractEnum;
import com.pig4cloud.pigx.strategy.enums.StrategyPositionContractEnum;
import com.pig4cloud.pigx.strategy.enums.StrategyStopOrderContractEnum;
import com.pig4cloud.pigx.strategy.enums.StrategySwitchContractEnum;
import com.pig4cloud.pigx.strategy.enums.StrategyTaskContractEnum;
import com.pig4cloud.pigx.strategy.executor.contract.order.StrategyOrderContractHandlerManager;
import com.pig4cloud.pigx.strategy.executor.contract.order.supporter.StrategyOrderContractCalculator;
import com.pig4cloud.pigx.strategy.executor.contract.stop_order.StrategyStopOrderContractHandlerManager;
import com.pig4cloud.pigx.strategy.executor.contract.task.supporter.StrategyTaskContractCalculator;
import com.pig4cloud.pigx.strategy.mapper.StrategyOrderContractMapper;
import com.pig4cloud.pigx.strategy.mapper.StrategyPositionContractMapper;
import com.pig4cloud.pigx.strategy.mapper.StrategyStopOrderContractMapper;
import com.pig4cloud.pigx.strategy.mapper.StrategySwitchContractMapper;
import com.pig4cloud.pigx.strategy.mapper.StrategyTaskContractMapper;
import com.pig4cloud.pigx.strategy.mapstruct.StrategyTaskContractStruct;
import com.pig4cloud.pigx.strategy.util.StrategyContractUtil;
import com.pig4cloud.pigx.system.constant.SystemPlatformConstant;
import com.pig4cloud.pigx.system.entity.SystemLog;
import com.pig4cloud.pigx.system.entity.SystemPlatform;
import com.pig4cloud.pigx.system.entity.SystemProduct;
import com.pig4cloud.pigx.system.enums.SystemLogEnum;
import com.pig4cloud.pigx.system.util.SystemLogUtil;
import com.pig4cloud.pigx.user.dto.TransactionThenDTO;
import com.pig4cloud.pigx.user.enums.FuelConsumeEnum;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.logging.log4j.util.Strings;
import org.redisson.api.RLock;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import javax.validation.constraints.NotNull;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

/**
 * 策略-合约处理器
 *
 * @since 2022/11/27 18:26
 */
@Slf4j
@Component
@EnableScheduling
@RequiredArgsConstructor
public class StrategyTaskContractHandler implements TransactionInterface {

    private final StrategySwitchContractMapper strategySwitchContractMapper;
    private final StrategyTaskContractMapper strategyTaskContractMapper;
    private final StrategyOrderContractMapper strategyOrderContractMapper;
    private final StrategyStopOrderContractMapper strategyStopOrderContractMapper;
    private final StrategyPositionContractMapper strategyPositionContractMapper;

    private final RedisUtils redisUtils;
    private final RedisTemplate redis;
    private final SecurityUtil securityUtil;
    private final MysqlUtil mysqlUtil;
    private final CommonUtil commonUtil;
    private final StrategyContractUtil strategyContractUtil;
    private final StrategyOrderContractHandlerManager strategyOrderContractHandlerManager;
    private final StrategyStopOrderContractHandlerManager strategyStopOrderContractHandlerManager;


    /**
     * 开启策略
     *
     * @param strategyId 策略id
     * @since 2022/11/28 10:30
     */
    public void open(@NotNull Long strategyId) {
        // System.err.println("启用策略");

        PigxUser user = SecurityUtil.getLoginUserNotNull();

        // 获取策略信息
        StrategyContract strategy = strategyContractUtil.getStrategyCache(strategyId, true);
        // 检查策略
        StrategyContractUtil.checkStrategyCache(strategy);

        // 获取策略组信息
        StrategyGroupContract group = strategyContractUtil.getStrategyGroupCache(strategy.getGroupId(), true);
        // 检查策略组
        StrategyContractUtil.checkStrategyGroupCache(group);

        // 查询燃油费账户
        BigDecimal fuelBalance = strategyContractUtil.getFuelBalance(user.getId());
        // 判断燃油费是否足够
        String strategyName = StrategyContractUtil.getStrategyName(group, strategy);
        if (fuelBalance.compareTo(group.getFuelFrozen()) < 0) {
            String errorMsg = String.format("策略 %s 无法启动，燃油币额度：%s，小于限制条件：%s", strategyName,
                    BigDecimalUtils.toStrRemoveExtraZero(fuelBalance),
                    BigDecimalUtils.toStrRemoveExtraZero(group.getFuelFrozen()));
            throw new DuckweedException(errorMsg);
        }

        try {
            // 判断USDT额度
            BigDecimal balance = strategyOrderContractHandlerManager.accountBalance(user.getId(), strategy.getPlatformId());
            // USDT额度不足
            if (balance.compareTo(strategy.getBalanceLimit()) < 0) {
                String errorMsg = String.format("策略 %s 无法启动，USDT额度：%s，小于限制条件：%s", strategyName,
                        BigDecimalUtils.toStrRemoveExtraZero(balance),
                        BigDecimalUtils.toStrRemoveExtraZero(strategy.getBalanceLimit()));
                throw new DuckweedException(errorMsg);
            }
        } catch (APIException e) {
            log.error(e.getMessage());
            throw new MqAckException("查询到USDT额度失败");
        }

        try {
            // 开始本地事务
            mysqlUtil.startLocalTx();

            // 生成开关
            StrategySwitchContract switchContract = buildSwitch(strategy.getPlatformId(), strategyId, user);
            // 生成任务
            StrategyTaskContract task = buildTask(switchContract.getPlatformId(), strategyId, user.getId(), user.getName(), switchContract.getId());

            // 提交本地事务
            mysqlUtil.commitLocalTx();

            // 缓存任务
            StrategyTaskContractCacheBO cache = StrategyTaskContractStruct.INSTANCE.entity2CacheBO(task);
            redisUtils.flushPersistentCache(RedisKeyEnum.STRATEGY_TASK_CONTRACT, task.getId(), cache);
            // 尝试立即执行任务
            run(task.getId());
            // 发送mq，策略开启邮件
            strategyContractUtil.sendEmailStrategyOpen(user.getId(), strategyName);
            // 发送mq，系统日志
            SystemLog systemLog = new SystemLog();
            securityUtil.initCreateAbout(systemLog, user, LocalDateTime.now());
            systemLog.setCreateName(user.getName());
            systemLog.setUpdateName(user.getName());
            systemLog.setOperatorType(SystemLogEnum.OperatorType.CLIENT.getValue());
            systemLog.setType(SystemLogEnum.Type.STRATEGY_OPEN);
            systemLog.setNewData(String.format("%s switchId：%s，taskId：%s", strategyName, task.getSwitchId(), task.getId()));
            strategyContractUtil.sendMqSystemLog(systemLog);

        } catch (Throwable e) {
            // 回滚本地事务
            mysqlUtil.rollbackLocalTx();
            throw e;
        }
    }

    /**
     * 任务执行完后，重新创建任务
     *
     * @param switchId 开关id
     * @since 2022/12/2 21:03
     */
    @RabbitListener(queues = StrategyMqConstant.QUEUE_FANOUT_STRATEGY_TASK_CONTRACT_CREATE)
    public void autoCreate(Long switchId) {
        // System.err.println("任务完成，重新创建任务");

        // 开启本地事务
        mysqlUtil.startLocalTx();

        // 获取开关信息并锁定
        QueryWrapper<StrategySwitchContract> querySwitch = new QueryWrapper<>();
        querySwitch.lambda()
                .eq(StrategySwitchContract::getId, switchId)
                .in(StrategySwitchContract::getStatus,
                        StrategySwitchContractEnum.Status.PAUSE.getValue(),
                        StrategySwitchContractEnum.Status.RUNNING.getValue())
                .last("for update");
        StrategySwitchContract switchContract = strategySwitchContractMapper.selectOne(querySwitch);
        // 开关已关闭、已结束、异常
        if (Objects.isNull(switchContract)) {
            // 回滚本地事务
            mysqlUtil.rollbackLocalTx();
            return;
        }
        String oldData = commonUtil.toJson(switchContract);

        StrategyContract strategy;
        StrategyGroupContract group;
        try {
            // 开关已暂停
            if (Objects.equals(StrategySwitchContractEnum.Status.PAUSE.getValue(), switchContract.getStatus())) {
                throw new DuckweedException("暂停后关闭");
            }

            // 获取策略信息
            strategy = strategyContractUtil.getStrategyCache(switchContract.getStrategyId(), true);
            // 检查策略
            StrategyContractUtil.checkStrategyCache(strategy);

            // 获取策略组信息
            group = strategyContractUtil.getStrategyGroupCache(strategy.getGroupId(), true);
            // 检查策略组
            StrategyContractUtil.checkStrategyGroupCache(group);

            // 查询燃油费账户
            BigDecimal fuelBalance = strategyContractUtil.getFuelBalance(switchContract.getUserId());
            String strategyName = StrategyContractUtil.getStrategyName(group, strategy);
            // 判断燃油费是否足够
            if (fuelBalance.compareTo(group.getFuelFrozen()) < 0) {
                String errorMsg = String.format("策略 %s 无法启动，燃油币额度：%s，小于限制条件：%s", strategyName,
                        BigDecimalUtils.toStrRemoveExtraZero(fuelBalance),
                        BigDecimalUtils.toStrRemoveExtraZero(group.getFuelFrozen()));
                throw new DuckweedException(errorMsg);
            }

            // 判断USDT额度
            BigDecimal balance = strategyOrderContractHandlerManager.accountBalance(switchContract.getUserId(), strategy.getPlatformId());
            // USDT额度不足
            if (balance.compareTo(strategy.getBalanceLimit()) < 0) {
                String errorMsg = String.format("策略 %s 无法启动，USDT额度：%s，小于限制条件：%s", strategyName,
                        BigDecimalUtils.toStrRemoveExtraZero(balance),
                        BigDecimalUtils.toStrRemoveExtraZero(strategy.getBalanceLimit()));
                throw new DuckweedException(errorMsg);
            }

        } catch (APIException e) {
            // 因为网络原因查询不到USDT额度
            log.error(e.getMessage());
            // 重新发送消息，防止只是个别用户的问题，阻塞到其它用户的任务
            strategyContractUtil.sendMqTaskCreate(switchId);
            return;
        } catch (Throwable e) {
            // 关闭开关
            switchContract.setRemark(e.getMessage());
            switchContract.setUpdateTime(LocalDateTime.now());
            switchContract.setUpdateBy(0L);
            switchContract.setUpdateName("系统");
            switchContract.setStatus(StrategySwitchContractEnum.Status.END.getValue());
            if (!switchContract.updateById()) {
                throw new DuckweedException("系统自动关闭开关失败");
            }

            // 提交本地事务
            mysqlUtil.commitLocalTx();
            // 发送mq，系统日志
            SystemLog systemLog = new SystemLog();
            securityUtil.initCreateAbout(systemLog, null, LocalDateTime.now());
            systemLog.setOperatorType(SystemLogEnum.OperatorType.SYSTEM.getValue());
            systemLog.setType(SystemLogEnum.Type.STRATEGY_CLOSE);
            systemLog.setOldData(oldData);
            systemLog.setNewData(commonUtil.toJson(switchContract));
            strategyContractUtil.sendMqSystemLog(systemLog);
            return;
        }

        try {
            // 生成任务
            StrategyTaskContract task = buildTask(switchContract.getPlatformId(),
                    switchContract.getStrategyId(),
                    switchContract.getUserId(),
                    "系统",
                    switchContract.getId());

            // 修改开关任务统计
            switchContract.setTaskCount(switchContract.getTaskCount() + 1);
            switchContract.setUpdateTime(LocalDateTime.now());
            switchContract.setUpdateBy(0L);
            switchContract.setUpdateName("系统");
            if (!switchContract.updateById()) {
                throw new DuckweedException("系统自动创建任务失败，修改开关状态失败");
            }

            // 提交本地事务
            mysqlUtil.commitLocalTx();

            // 缓存任务
            StrategyTaskContractCacheBO cache = StrategyTaskContractStruct.INSTANCE.entity2CacheBO(task);
            redisUtils.flushPersistentCache(RedisKeyEnum.STRATEGY_TASK_CONTRACT, task.getId(), cache);
            // 尝试立即执行任务
            run(task.getId());
            // 发送mq，系统日志
            SystemLog systemLog = new SystemLog();
            securityUtil.initCreateAbout(systemLog, null, LocalDateTime.now());
            systemLog.setOperatorType(SystemLogEnum.OperatorType.SYSTEM.getValue());
            systemLog.setType(SystemLogEnum.Type.STRATEGY_AUTO_CREATE);
            systemLog.setOldData(oldData);
            String strategyName = StrategyContractUtil.getStrategyName(group, strategy);
            systemLog.setNewData(String.format("%s switchId：%s，taskId：%s", strategyName, task.getSwitchId(), task.getId()));
            strategyContractUtil.sendMqSystemLog(systemLog);

        } catch (Throwable e) {
            // 回滚本地事务
            mysqlUtil.rollbackLocalTx();
            log.error(e.getMessage());
            throw e;
        }
    }

    /**
     * 生成任务
     *
     * @param platformId 平台id
     * @param strategyId 策略id
     * @param userId     用户id
     * @param username   用户名称
     * @param switchId   开关id
     * @return {@link StrategyTaskContract}
     * @since 2022/11/28 14:41
     */
    private StrategyTaskContract buildTask(Long platformId, Long strategyId, Long userId, String username, Long switchId) {
        StrategyTaskContract task = strategyContractUtil.initTask(platformId, strategyId, userId, username, switchId);
        if (strategyTaskContractMapper.add(task) != 1) {
            throw new DuckweedException("操作失败");
        }

        return task;
    }

    /**
     * 生成开关
     *
     * @param platformId 平台id
     * @param strategyId 策略id
     * @param user       用户
     * @return {@link StrategySwitchContract}
     * @since 2022/11/28 11:47
     */
    private StrategySwitchContract buildSwitch(Long platformId, Long strategyId, PigxUser user) {
        // 初始化开关信息
        StrategySwitchContract switchContract = new StrategySwitchContract();
        switchContract.setPlatformId(platformId);
        switchContract.setUserId(user.getId());
        switchContract.setStrategyId(strategyId);
        switchContract.setProfit(BigDecimal.ZERO);
        switchContract.setNetProfit(BigDecimal.ZERO);
        switchContract.setFuelCost(BigDecimal.ZERO);
        switchContract.setErrorSettlementProfit(BigDecimal.ZERO);
        switchContract.setTaskCount(1);
        switchContract.setSendFuelReminder(CommonDictEnum.YesOrNo.NO.getValue());
        switchContract.setStatus(StrategySwitchContractEnum.Status.RUNNING.getValue());
        switchContract.setCreateName(user.getName());
        switchContract.setUpdateName(user.getName());
        securityUtil.initCreateAbout(switchContract);

        // 尝试新增开关
        if (strategySwitchContractMapper.add(switchContract) != 1) {
            throw new DuckweedException("存在运行中/异常的策略");
        }

        return switchContract;
    }

    /**
     * 分发任务
     *
     * @since 2022/11/28 10:32
     */
    @Scheduled(fixedDelay = StrategyScheduleConstant.INTERNAL_TASK_RUN_DISPATCHER)
    public void dispatch() {
        RLock lock = redisUtils.getLock(StrategyRedisLockConstant.TASK_RUN_DISPATCH, StrategyRedisLockConstant.TASK_RUN_DISPATCH_EXPIRE);
        if (Objects.isNull(lock)) {
            return;
        }

        try {
            Set<String> keys = redis.opsForHash().keys(RedisKeyEnum.STRATEGY_TASK_CONTRACT.getPersistentCacheKey());
            // if (keys.size() > 0) {
            //     System.err.println("分发任务" + keys.size() + "个");
            // }

            // 发送mq，运行任务
            keys.forEach(strategyContractUtil::sendMqRunTask);
        } finally {
            redisUtils.releaseLock(lock);
        }
    }

    /**
     * 运行任务
     *
     * @param taskId 任务id
     * @since 2022/11/28 10:32
     */
    @RabbitListener(queues = StrategyMqConstant.EXCHANGE_FANOUT_STRATEGY_TASK_CONTRACT_HANDLE)
    public void run(Long taskId) {
        // long long1 = System.currentTimeMillis();

        // System.err.println("运行任务：" + taskId);
        RLock lock = null;
        try {
            StrategyTaskContractCacheBO task = strategyContractUtil.getStrategyTaskCache(taskId, false);
            // 获取分布式锁
            String key = String.format(StrategyRedisLockConstant.TASK_RUN, task.getSwitchId());
            lock = redisUtils.getLock(key, null);
            if (Objects.isNull(lock)) {
                // System.err.println("运行任务时，未能获取到任务分布式锁");
                return;
            }

            TaskRunner taskRunner = new TaskRunner(redisUtils,
                    strategyOrderContractHandlerManager,
                    taskId,
                    true,
                    true);

            // taskRunner在执行了下单逻辑后，会返回true
            // 开仓
            if (taskRunner.openPosition()) {
                return;
            }

            // 检查第三方止盈止损是否触发
            if (taskRunner.checkExternalStopProfitLost()) {
                return;
            }

            // 止盈止损
            if (taskRunner.stopProfitLost()) {
                return;
            }
            taskRunner.flushCache();

            // 补仓锁仓
            if (taskRunner.addPosition()) {
                return;
            }
            taskRunner.flushCache();

        } catch (MqAckLogException e) {
            log.error(e.getMessage());
        } catch (MqAckException ignore) {
        } catch (Throwable e) {
            e.printStackTrace();
        } finally {
            // 释放分布式锁
            redisUtils.releaseLock(lock);
        }

        // System.err.println("一次任务处理耗时：" + (System.currentTimeMillis() - long1));
    }

    /**
     * 暂停开关
     * <p>mysql锁定顺序： 开关 -> 任务 -> 订单 -> 仓位
     *
     * @param bo bo
     * @since 2022/11/28 10:32
     */
    public void pauseWhenLowFuel(SyncOrderBO bo) {
        String msgPrefix = "策略暂停-";
        LocalDateTime now = LocalDateTime.now();
        Long userId = 0L;
        String username = "系统";
        String remark = "燃油币账户额度不足";

        StrategySwitchContract switchContract = bo.getSwitchContract();
        // 只暂停运行中的策略
        if (!Objects.equals(StrategySwitchContractEnum.Status.RUNNING.getValue(), switchContract.getStatus())) {
            return;
        }
        bo.setHasPauseWhenLowFuel(true);

        String oldData = commonUtil.toJson(switchContract);
        switchContract.setRemark(remark);
        switchContract.setStatus(StrategySwitchContractEnum.Status.PAUSE.getValue());
        switchContract.setUpdateTime(now);
        switchContract.setUpdateBy(userId);
        switchContract.setUpdateName(username);
        if (!switchContract.updateById()) {
            throw new MqNotAckException(msgPrefix + "暂停策略失败，switchId：" + switchContract.getId());
        }

        StrategyTaskContract task = bo.getTask();
        // 修改任务
        task.setRemark(remark);
        task.setStatus(StrategyTaskContractEnum.Status.PAUSE.getValue());
        task.setUpdateTime(now);
        task.setUpdateBy(userId);
        task.setUpdateName(username);
        if (strategyTaskContractMapper.updateById(task) != 1) {
            throw new MqNotAckException(msgPrefix + "暂停任务失败，taskId：" + task.getId());
        }

        // 记录日志参数，任务完成后发送
        SystemLog systemLog = new SystemLog();
        systemLog.setLogId(null);
        securityUtil.initCreateAbout(systemLog, null, now);
        systemLog.setCreateName(username);
        systemLog.setUpdateName(username);
        systemLog.setOperatorType(SystemLogEnum.OperatorType.SYSTEM.getValue());
        systemLog.setType(SystemLogEnum.Type.STRATEGY_PAUSE);
        systemLog.setOldData(oldData);
        systemLog.setNewData(commonUtil.toJson(switchContract));
        bo.getSystemLogList().add(systemLog);
    }

    /**
     * 暂停开关
     * <p>mysql锁定顺序： 开关 -> 任务 -> 订单 -> 仓位
     *
     * @param dto dto
     * @since 2022/11/28 10:32
     */
    @Transactional(rollbackFor = Throwable.class)
    public void pause(StrategyContractPauseDTO dto) {
        // 获取分布式锁
        RLock taskLock = null;
        if (dto.isNeedRedisLock()) {
            String key = String.format(StrategyRedisLockConstant.TASK_RUN, dto.getSwitchId());
            taskLock = redisUtils.waitLock(key, null);
        }

        String msgPrefix = "策略暂停-";
        LocalDateTime now = LocalDateTime.now();
        PigxUser user = null;
        Long userId;
        String username;
        if (dto.isSystem()) {
            userId = 0L;
            username = "系统";
        } else {
            user = SecurityUtil.getLoginUserNotNull();
            userId = user.getId();
            username = user.getName();
        }

        try {
            // 查询并锁定开关
            QueryWrapper<StrategySwitchContract> querySwitch = new QueryWrapper<>();
            querySwitch.lambda()
                    .eq(StrategySwitchContract::getId, dto.getSwitchId())
                    .eq(StrategySwitchContract::getStatus, StrategySwitchContractEnum.Status.RUNNING.getValue())
                    .last("for update");
            StrategySwitchContract switchContract = strategySwitchContractMapper.selectOne(querySwitch);
            if (Objects.isNull(switchContract)) {
                throw new MqAckException(msgPrefix + "状态已发生改变，switchId：" + dto.getSwitchId());
            }
            String oldData = commonUtil.toJson(switchContract);

            switchContract.setRemark(dto.getRemark());
            switchContract.setStatus(StrategySwitchContractEnum.Status.PAUSE.getValue());
            switchContract.setUpdateTime(now);
            switchContract.setUpdateBy(userId);
            switchContract.setUpdateName(username);
            if (!switchContract.updateById()) {
                throw new MqNotAckException(msgPrefix + "暂停策略失败，switchId：" + dto.getSwitchId());
            }

            // 查询运行中的任务，并锁定
            QueryWrapper<StrategyTaskContract> queryTask = new QueryWrapper<>();
            queryTask.lambda()
                    .eq(StrategyTaskContract::getSwitchId, dto.getSwitchId())
                    .eq(StrategyTaskContract::getStatus, StrategyTaskContractEnum.Status.RUNNING.getValue())
                    .last("for update");
            StrategyTaskContract task = strategyTaskContractMapper.selectOne(queryTask);

            boolean hasTask = !Objects.isNull(task);
            if (hasTask) {
                // 修改任务
                task.setRemark(dto.getRemark());
                task.setStatus(StrategyTaskContractEnum.Status.PAUSE.getValue());
                task.setUpdateTime(now);
                task.setUpdateBy(userId);
                task.setUpdateName(username);
                if (strategyTaskContractMapper.updateById(task) != 1) {
                    throw new MqNotAckException(msgPrefix + "暂停任务失败，taskId：" + task.getId());
                }

                // 获取任务缓存
                StrategyTaskContractCacheBO taskCache = strategyContractUtil.getStrategyTaskCache(task.getId(), false);
                // 刷新任务缓存
                strategyContractUtil.flushTaskCacheByEntity(taskCache, task, taskCache.getLastOpenAvg(), null);
            }

            // 获取策略、策略组缓存
            StrategyContract strategy = strategyContractUtil.getStrategyCache(switchContract.getStrategyId(), false);
            StrategyGroupContract group = strategyContractUtil.getStrategyGroupCache(strategy.getGroupId(), false);
            // 发送mq，策略暂停邮件
            strategyContractUtil.sendEmailStrategyPause(switchContract.getUserId(), strategy, group, dto.getRemark());
            // 发送mq，系统日志
            SystemLog systemLog = new SystemLog();
            // 批量操作记录logId
            systemLog.setLogId(dto.getLogId());
            securityUtil.initCreateAbout(systemLog, user, now);
            systemLog.setCreateName(username);
            systemLog.setUpdateName(username);
            systemLog.setOperatorType(dto.isSystem() ? SystemLogEnum.OperatorType.SYSTEM.getValue() : SystemLogEnum.OperatorType.MANAGER.getValue());
            systemLog.setType(SystemLogEnum.Type.STRATEGY_PAUSE);
            systemLog.setOldData(oldData);
            systemLog.setNewData(commonUtil.toJson(switchContract));
            strategyContractUtil.sendMqSystemLog(systemLog);

        } catch (Throwable e) {
            // 发送mq，系统日志
            SystemLog systemLog = new SystemLog();
            // 批量操作记录logId
            systemLog.setLogId(dto.getLogId());
            securityUtil.initCreateAbout(systemLog, user, now);
            systemLog.setCreateName(username);
            systemLog.setUpdateName(username);
            systemLog.setOperatorType(dto.isSystem() ? SystemLogEnum.OperatorType.SYSTEM.getValue() : SystemLogEnum.OperatorType.MANAGER.getValue());
            systemLog.setType(SystemLogEnum.Type.STRATEGY_PAUSE);
            systemLog.setOldData(null);
            systemLog.setNewData(String.format("交易员暂停策略失败，switchId：%s，原因：%s",
                    dto.getSwitchId(),
                    e.getMessage()));
            strategyContractUtil.sendMqSystemLog(systemLog);
            throw e;
        } finally {
            redisUtils.releaseLock(taskLock);
        }
    }

    /**
     * mq消息暂停开关
     *
     * @param dto dto
     * @since 2022/11/28 10:32
     */
    @RabbitListener(queues = StrategyMqConstant.QUEUE_FANOUT_STRATEGY_CONTRACT_PAUSE)
    public void pauseMq(StrategyContractPauseDTO dto) {

        try {
            pause(dto);
        } catch (MqAckLogException e) {
            log.error(e.getMessage());
        } catch (MqAckException ignore) {

        }

    }

    /**
     * 恢复开关
     * <p>mysql锁定顺序： 开关 -> 任务 -> 订单 -> 仓位
     *
     * @param dto dto
     * @since 2022/11/28 10:32
     */
    @Transactional(rollbackFor = Throwable.class)
    public void restore(StrategyContractRestoreDTO dto) {
        // 获取分布式锁
        String key = String.format(StrategyRedisLockConstant.TASK_RUN, dto.getSwitchId());
        RLock taskLock = redisUtils.waitLock(key, null);

        String msgPrefix = "策略恢复-";
        LocalDateTime now = LocalDateTime.now();
        PigxUser user = SecurityUtil.getLoginUser();
        Long userId;
        String username;
        if (Objects.isNull(user)) {
            userId = 0L;
            username = "系统";
        } else {
            userId = user.getId();
            username = user.getName();
        }

        try {
            // 查询并锁定开关
            QueryWrapper<StrategySwitchContract> querySwitch = new QueryWrapper<>();
            querySwitch.lambda()
                    .eq(StrategySwitchContract::getId, dto.getSwitchId())
                    .eq(StrategySwitchContract::getStatus, StrategySwitchContractEnum.Status.PAUSE.getValue())
                    .last("for update");
            StrategySwitchContract switchContract = strategySwitchContractMapper.selectOne(querySwitch);
            if (Objects.isNull(switchContract)) {
                throw new MqAckException(msgPrefix + "状态已发生改变，switchId：" + dto.getSwitchId());
            }
            // 旧数据
            String oldData = commonUtil.toJson(switchContract);

            switchContract.setRemark(dto.getRemark());
            switchContract.setStatus(StrategySwitchContractEnum.Status.RUNNING.getValue());
            switchContract.setUpdateTime(now);
            switchContract.setUpdateBy(userId);
            switchContract.setUpdateName(username);
            if (!switchContract.updateById()) {
                throw new MqNotAckException(msgPrefix + "恢复策略失败，switchId：" + dto.getSwitchId());
            }

            // 查询暂停的任务，并锁定
            QueryWrapper<StrategyTaskContract> queryTask = new QueryWrapper<>();
            queryTask.lambda()
                    .eq(StrategyTaskContract::getSwitchId, dto.getSwitchId())
                    .eq(StrategyTaskContract::getStatus, StrategyTaskContractEnum.Status.PAUSE.getValue())
                    .last("for update");
            StrategyTaskContract task = strategyTaskContractMapper.selectOne(queryTask);

            boolean hasTask = !Objects.isNull(task);
            if (hasTask) {
                // 修改任务
                task.setRemark(dto.getRemark());
                task.setStatus(StrategyTaskContractEnum.Status.RUNNING.getValue());
                task.setUpdateTime(now);
                task.setUpdateBy(userId);
                task.setUpdateName(username);
                if (strategyTaskContractMapper.updateById(task) != 1) {
                    throw new MqNotAckException(msgPrefix + "恢复任务失败，switchId：" + dto.getSwitchId());
                }

                // 获取任务缓存
                StrategyTaskContractCacheBO taskCache = strategyContractUtil.getStrategyTaskCache(task.getId(), false);
                // 刷新任务缓存
                strategyContractUtil.flushTaskCacheByEntity(taskCache, task, taskCache.getLastOpenAvg(), null);
            }

            // 获取策略、策略组缓存
            StrategyContract strategy = strategyContractUtil.getStrategyCache(switchContract.getStrategyId(), false);
            StrategyGroupContract group = strategyContractUtil.getStrategyGroupCache(strategy.getGroupId(), false);
            // 发送mq，策略恢复邮件
            strategyContractUtil.sendEmailStrategyRestore(switchContract.getUserId(), strategy, group, dto.getRemark());
            // 发送mq，系统日志
            SystemLog systemLog = new SystemLog();
            // 批量操作记录logId
            systemLog.setLogId(dto.getLogId());
            securityUtil.initCreateAbout(systemLog, user, now);
            systemLog.setCreateName(username);
            systemLog.setUpdateName(username);
            systemLog.setOperatorType(SystemLogUtil.getOperatorTypeByUser(user, false));
            systemLog.setType(SystemLogEnum.Type.STRATEGY_RESTORE);
            systemLog.setOldData(oldData);
            systemLog.setNewData(commonUtil.toJson(switchContract));
            strategyContractUtil.sendMqSystemLog(systemLog);

        } catch (Throwable e) {
            // 发送mq，系统日志
            SystemLog systemLog = new SystemLog();
            // 批量操作记录logId
            systemLog.setLogId(dto.getLogId());
            securityUtil.initCreateAbout(systemLog, user, now);
            systemLog.setCreateName(username);
            systemLog.setUpdateName(username);
            systemLog.setOperatorType(SystemLogUtil.getOperatorTypeByUser(user, false));
            systemLog.setType(SystemLogEnum.Type.STRATEGY_RESTORE);
            systemLog.setOldData(null);
            systemLog.setNewData(String.format("交易员恢复策略失败，switchId：%s，原因：%s",
                    dto.getSwitchId(),
                    e.getMessage()));
            strategyContractUtil.sendMqSystemLog(systemLog);
            throw e;

        } finally {
            redisUtils.releaseLock(taskLock);
        }
    }

    /**
     * mq消息恢复开关
     *
     * @param dto dto
     * @since 2022/11/28 10:32
     */
    @RabbitListener(queues = StrategyMqConstant.QUEUE_FANOUT_STRATEGY_CONTRACT_RESTORE)
    public void restoreMq(StrategyContractRestoreDTO dto) {

        try {
            restore(dto);
        } catch (MqAckLogException e) {
            log.error(e.getMessage());
        } catch (MqAckException ignore) {

        }

    }

    /**
     * 平仓
     * <p>mysql锁定顺序： 开关 -> 任务 -> 订单 -> 仓位
     *
     * @param dto dto
     * @since 2022/11/28 10:32
     */
    public void positionClose(StrategyPositionContractCloseDTO dto) {
        RLock lock = null;
        PigxUser user = SecurityUtil.getLoginUser();
        LocalDateTime now = LocalDateTime.now();
        String username = Objects.isNull(user) ? "系统" : user.getName();

        try {
            StrategyTaskContractCacheBO task = strategyContractUtil.getStrategyTaskCache(dto.getTaskId(), false);
            // 获取分布式锁
            String key = String.format(StrategyRedisLockConstant.TASK_RUN, task.getSwitchId());
            lock = redisUtils.waitLock(key, null);

            TaskRunner taskRunner = new TaskRunner(redisUtils, strategyOrderContractHandlerManager, dto.getTaskId(), true, false);
            List<PlaceOrderBO> paramList = taskRunner.buildCloseManualParam();
            // 下单
            strategyOrderContractHandlerManager.placeOrder(paramList);

            // 发送mq，系统日志
            SystemLog systemLog = new SystemLog();
            // 批量操作记录logId
            systemLog.setLogId(dto.getLogId());
            securityUtil.initCreateAbout(systemLog, user, now);
            systemLog.setCreateName(username);
            systemLog.setUpdateName(username);
            systemLog.setOperatorType(SystemLogUtil.getOperatorTypeByUser(user, false));
            systemLog.setType(SystemLogEnum.Type.STRATEGY_POSITION_CLOSE);
            systemLog.setOldData(null);
            systemLog.setNewData(String.format("交易员平仓，taskId：%s", dto.getTaskId()));
            strategyContractUtil.sendMqSystemLog(systemLog);

        } catch (Throwable e) {
            e.printStackTrace();
            // 发送mq，系统日志
            SystemLog systemLog = new SystemLog();
            // 批量操作记录logId
            systemLog.setLogId(dto.getLogId());
            securityUtil.initCreateAbout(systemLog, user, now);
            systemLog.setCreateName(username);
            systemLog.setUpdateName(username);
            systemLog.setOperatorType(SystemLogUtil.getOperatorTypeByUser(user, false));
            systemLog.setType(SystemLogEnum.Type.STRATEGY_RESTORE);
            systemLog.setOldData(null);
            systemLog.setNewData(String.format("交易员平仓失败，taskId：%s，原因：%s",
                    dto.getTaskId(),
                    e.getMessage()));
            strategyContractUtil.sendMqSystemLog(systemLog);
            throw new MqAckLogException("平仓失败");

        } finally {
            redisUtils.releaseLock(lock);
        }
    }

    /**
     * mq消息平仓
     *
     * @param dto dto
     * @since 2022/11/28 10:32
     */
    @RabbitListener(queues = StrategyMqConstant.QUEUE_FANOUT_STRATEGY_POSITION_CONTRACT_CLOSE)
    public void positionCloseMq(StrategyPositionContractCloseDTO dto) {
        try {
            positionClose(dto);
        } catch (MqAckLogException e) {
            log.error(e.getMessage());
        } catch (MqAckException ignore) {

        }
    }

    /**
     * 手动开仓
     *
     * @param dto 入参
     * @since 2023/1/17 14:02
     */
    public void positionOpen(StrategyPositionContractOpenDTO dto) {
        RLock lock = null;
        PigxUser user = SecurityUtil.getLoginUser();
        LocalDateTime now = LocalDateTime.now();
        String username = Objects.isNull(user) ? "系统" : user.getName();

        try {
            // 获取分布式锁
            StrategyTaskContractCacheBO taskCache = strategyContractUtil.getStrategyTaskCache(dto.getTaskId(), false);
            String key = String.format(StrategyRedisLockConstant.TASK_RUN, taskCache.getSwitchId());
            lock = redisUtils.waitLock(key, null);

            // 下单
            TaskRunner taskRunner = new TaskRunner(redisUtils, strategyOrderContractHandlerManager, dto.getTaskId(), true, false);
            PlaceOrderBO param = taskRunner.buildOpenManualParam(dto.getSide(), dto.getNum());
            strategyOrderContractHandlerManager.placeOrder(Collections.singletonList(param));

            // 发送mq，系统日志
            SystemLog systemLog = new SystemLog();
            securityUtil.initCreateAbout(systemLog, user, now);
            systemLog.setCreateName(username);
            systemLog.setUpdateName(username);
            systemLog.setOperatorType(SystemLogUtil.getOperatorTypeByUser(user, false));
            systemLog.setType(SystemLogEnum.Type.STRATEGY_POSITION_OPEN_MANUAL);
            systemLog.setOldData(null);
            systemLog.setNewData(String.format("交易员手动开仓，taskId：%s，side：%s，num:%s",
                    dto.getTaskId(),
                    dto.getSide(),
                    dto.getNum()));
            strategyContractUtil.sendMqSystemLog(systemLog);

        } catch (Throwable e) {
            e.printStackTrace();

            // 发送mq，系统日志
            SystemLog systemLog = new SystemLog();
            securityUtil.initCreateAbout(systemLog, user, now);
            systemLog.setCreateName(username);
            systemLog.setUpdateName(username);
            systemLog.setOperatorType(SystemLogUtil.getOperatorTypeByUser(user, false));
            systemLog.setType(SystemLogEnum.Type.STRATEGY_POSITION_OPEN_MANUAL);
            systemLog.setOldData(null);
            systemLog.setNewData(String.format("交易员手动开仓失败，taskId：%s，side：%s，num:%s，原因：%s",
                    dto.getTaskId(),
                    dto.getSide(),
                    dto.getNum(),
                    e.getMessage()));
            strategyContractUtil.sendMqSystemLog(systemLog);

            throw new MqAckLogException("手动开仓失败，原因：" + e.getMessage());
        } finally {
            redisUtils.releaseLock(lock);
        }
    }

    /**
     * 手动开仓
     *
     * @param dto dto
     * @since 2023/2/12 20:34
     */
    @RabbitListener(queues = StrategyMqConstant.EXCHANGE_FANOUT_STRATEGY_POSITION_CONTRACT_OPEN_MANUAL)
    public void positionOpenMq(StrategyPositionContractOpenDTO dto) {
        try {
            positionOpen(dto);
        } catch (MqAckLogException e) {
            log.error(e.getMessage());
        } catch (MqAckException ignore) {

        }
    }

    /**
     * 平部分仓位
     *
     * @param dto dto
     * @since 2022/11/28 10:32
     */
    public void positionClosePart(StrategyPositionContractClosePartDTO dto) {
        RLock lock = null;
        PigxUser user = SecurityUtil.getLoginUser();
        LocalDateTime now = LocalDateTime.now();
        String username = Objects.isNull(user) ? "系统" : user.getName();

        try {
            // 获取分布式锁
            StrategyTaskContractCacheBO taskCache = strategyContractUtil.getStrategyTaskCache(dto.getTaskId(), false);
            String key = String.format(StrategyRedisLockConstant.TASK_RUN, taskCache.getSwitchId());
            lock = redisUtils.waitLock(key, null);

            // 下单
            TaskRunner taskRunner = new TaskRunner(redisUtils, strategyOrderContractHandlerManager, dto.getTaskId(), true, false);
            PlaceOrderBO param = taskRunner.buildClosePartParam(dto.getSide(), dto.getNum());
            strategyOrderContractHandlerManager.placeOrder(Collections.singletonList(param));

            // 发送mq，系统日志
            SystemLog systemLog = new SystemLog();
            securityUtil.initCreateAbout(systemLog, user, now);
            systemLog.setCreateName(username);
            systemLog.setUpdateName(username);
            systemLog.setOperatorType(SystemLogUtil.getOperatorTypeByUser(user, false));
            systemLog.setType(SystemLogEnum.Type.STRATEGY_POSITION_CLOSE_SINGLE);
            systemLog.setOldData(null);
            systemLog.setNewData(String.format("交易员部分平仓，taskId：%s，side：%s",
                    dto.getTaskId(),
                    dto.getSide()));
            strategyContractUtil.sendMqSystemLog(systemLog);

        } catch (Throwable e) {
            e.printStackTrace();

            // 发送mq，系统日志
            SystemLog systemLog = new SystemLog();
            securityUtil.initCreateAbout(systemLog, user, now);
            systemLog.setCreateName(username);
            systemLog.setUpdateName(username);
            systemLog.setOperatorType(SystemLogUtil.getOperatorTypeByUser(user, false));
            systemLog.setType(SystemLogEnum.Type.STRATEGY_POSITION_CLOSE_SINGLE);
            systemLog.setOldData(null);
            systemLog.setNewData(String.format("交易员部分平仓失败，taskId：%s，side：%s，原因：%s",
                    dto.getTaskId(),
                    dto.getSide(),
                    e.getMessage()));
            strategyContractUtil.sendMqSystemLog(systemLog);
            throw new MqAckLogException("平仓失败，原因：" + e.getMessage());
        } finally {
            redisUtils.releaseLock(lock);
        }
    }

    /**
     * 部分平仓
     *
     * @param dto dto
     * @since 2023/2/12 20:34
     */
    @RabbitListener(queues = StrategyMqConstant.QUEUE_FANOUT_STRATEGY_POSITION_CONTRACT_CLOSE_PART)
    public void positionClosePartMq(StrategyPositionContractClosePartDTO dto) {
        try {
            positionClosePart(dto);
        } catch (MqAckLogException e) {
            log.error(e.getMessage());
        } catch (MqAckException ignore) {

        }
    }

    /**
     * mq消息用户充值燃油币后额度大于0
     *
     * @param userId 用户id
     * @since 2022/11/28 10:32
     */
    @RabbitListener(queues = StrategyMqConstant.QUEUE_FANOUT_STRATEGY_CONTRACT_FUEL_RECHARGE)
    public void restoreWhenFuelRecharge(Long userId) {

        // 查询用户因为燃油币不足被暂停的开关
        QueryWrapper<StrategySwitchContract> querySwitch = new QueryWrapper<>();
        querySwitch.lambda()
                .select(StrategySwitchContract::getId)
                .eq(StrategySwitchContract::getUserId, userId)
                .eq(StrategySwitchContract::getRemark, "燃油币账户额度不足")
                .eq(StrategySwitchContract::getStatus, StrategySwitchContractEnum.Status.PAUSE.getValue());
        StrategySwitchContract switchContract = strategySwitchContractMapper.selectOne(querySwitch);
        if (Objects.isNull(switchContract)) {
            // 发送余额告警检查消息
            strategyContractUtil.sendMqFuelWarning(userId);
            return;
        }

        StrategyContractRestoreDTO dto = new StrategyContractRestoreDTO();
        dto.setRemark("用户充值燃油币后恢复因燃油币账户额度不足而暂停的策略为运行状态");
        dto.setSwitchId(switchContract.getId());
        try {
            restore(dto);
        } catch (MqAckLogException e) {
            log.error(e.getMessage());
        } catch (MqAckException ignore) {
        }

        // 发送余额告警检查消息
        strategyContractUtil.sendMqFuelWarning(userId);
    }

    /**
     * 关闭策略
     * <p>mysql锁定顺序： 开关 -> 任务 -> 订单 -> 仓位
     *
     * @param dto dto
     * @since 2022/11/28 10:33
     */
    public void close(StrategyContractCloseDTO dto) {
        // 获取分布式锁
        RLock taskLock = null;
        if (dto.isNeedRedisLock()) {
            String key = String.format(StrategyRedisLockConstant.TASK_RUN, dto.getSwitchId());
            taskLock = redisUtils.waitLock(key, null);
        }

        LocalDateTime now = LocalDateTime.now();
        PigxUser user = null;
        Long userId;
        String username;
        if (dto.isSystem()) {
            userId = 0L;
            username = "系统";
        } else {
            user = SecurityUtil.getLoginUserNotNull();
            userId = user.getId();
            username = user.getName();
        }

        try {
            // 查询并锁定开关
            QueryWrapper<StrategySwitchContract> querySwitch = new QueryWrapper<>();
            querySwitch.lambda()
                    .eq(StrategySwitchContract::getId, dto.getSwitchId())
                    .in(StrategySwitchContract::getStatus,
                            StrategySwitchContractEnum.Status.PAUSE.getValue(),
                            StrategySwitchContractEnum.Status.ERROR.getValue(),
                            StrategySwitchContractEnum.Status.RUNNING.getValue())
                    .last("for update");
            StrategySwitchContract switchContract = strategySwitchContractMapper.selectOne(querySwitch);
            if (Objects.isNull(switchContract)) {
                throw new MqAckException("状态已发生改变");
            }
            String oldData = commonUtil.toJson(switchContract);

            switchContract.setRemark(dto.getRemark());
            switchContract.setStatus(StrategySwitchContractEnum.Status.CLOSE.getValue());
            switchContract.setUpdateTime(now);
            switchContract.setUpdateBy(userId);
            switchContract.setUpdateName(username);
            if (!switchContract.updateById()) {
                throw new MqNotAckException("关闭策略失败");
            }

            // 查询运行中的任务，并锁定
            QueryWrapper<StrategyTaskContract> queryTask = new QueryWrapper<>();
            queryTask.lambda()
                    .eq(StrategyTaskContract::getSwitchId, dto.getSwitchId())
                    .in(StrategyTaskContract::getStatus,
                            StrategyTaskContractEnum.Status.PAUSE.getValue(),
                            StrategyTaskContractEnum.Status.ERROR.getValue(),
                            StrategyTaskContractEnum.Status.RUNNING.getValue())
                    .last("for update");
            StrategyTaskContract task = strategyTaskContractMapper.selectOne(queryTask);

            boolean hasTask = !Objects.isNull(task);
            if (hasTask) {
                if (task.getPendingOrderNum() > 0) {
                    throw new MqAckException("策略正在处理订单，请稍后再试");
                }
                if (Objects.equals(StrategyTaskContractEnum.ErrorSettlementStatus.RUNNING.getValue(), task.getErrorSettlementStatus())) {
                    throw new MqAckException("当前任务正在进行持仓异常部分的利润结算，请稍后再试");
                }

                // 修改任务
                task.setRemark(dto.getRemark());
                // 是app用户操作的关闭 && 还未处理异常结算
                boolean notErrorSettlement = Objects.equals(StrategyTaskContractEnum.ErrorSettlementStatus.NONE.getValue(), task.getErrorSettlementStatus());
                if (!dto.isSystem() && notErrorSettlement) {
                    // 获取策略信息
                    StrategyContract strategy = strategyContractUtil.getStrategyCache(task.getStrategyId(), false);
                    // 查询产品
                    SystemProduct product = strategyContractUtil.getProductById(strategy.getProductId(), false);
                    // 查询价格
                    BigDecimal symbolPrice = null;
                    if (product.getPlatformId().equals(SystemPlatformConstant.OKX)) {
                        symbolPrice = redisUtils.getSymbolPrice(strategy.getSymbolName(), CommonRedisConstants.OKX + CommonRedisConstants.LAST_PRICE);
                    }
                    task.setErrorSettlementPrice(symbolPrice);
                    task.setErrorSettlementStatus(StrategyTaskContractEnum.ErrorSettlementStatus.RUNNING.getValue());
                }
                task.setStatus(StrategyTaskContractEnum.Status.CLOSE.getValue());
                task.setUpdateTime(now);
                task.setUpdateBy(userId);
                task.setUpdateName(username);
                if (strategyTaskContractMapper.updateById(task) != 1) {
                    throw new MqNotAckException("关闭任务失败");
                }

                // app用户关闭，如果有运行中的任务，结算持仓异常部分的利润
                if (!dto.isSystem() && notErrorSettlement) {
                    StrategyContractErrorSettlementDTO errorSettlementDTO = new StrategyContractErrorSettlementDTO();
                    errorSettlementDTO.setTaskId(task.getId());
                    errorSettlementDTO.setStatus(task.getStatus());
                    errorSettlement(errorSettlementDTO, task);
                }

                // 移除任务缓存
                redisUtils.delPersistentCache(RedisKeyEnum.STRATEGY_TASK_CONTRACT, task.getId());
            }

            // 获取策略、策略组缓存
            StrategyContract strategy = strategyContractUtil.getStrategyCache(switchContract.getStrategyId(), false);
            StrategyGroupContract group = strategyContractUtil.getStrategyGroupCache(strategy.getGroupId(), false);
            // 发送mq，策略关闭邮件
            String strategyName = StrategyContractUtil.getStrategyName(group, strategy);
            strategyContractUtil.sendEmailStrategyClose(switchContract.getUserId(), strategyName, dto.getRemark());
            // 发送mq，系统日志
            SystemLog systemLog = new SystemLog();
            securityUtil.initCreateAbout(systemLog, user, now);
            systemLog.setCreateName(username);
            systemLog.setUpdateName(username);
            systemLog.setOperatorType(dto.isSystem() ? SystemLogEnum.OperatorType.SYSTEM.getValue() : SystemLogEnum.OperatorType.CLIENT.getValue());
            systemLog.setType(SystemLogEnum.Type.STRATEGY_CLOSE);
            systemLog.setOldData(oldData);
            systemLog.setNewData(commonUtil.toJson(switchContract));
            strategyContractUtil.sendMqSystemLog(systemLog);
        } catch (Throwable e) {
            SeataUtils.rollbackGlobalTx();
            e.printStackTrace();
            throw e;
        } finally {
            redisUtils.releaseLock(taskLock);
        }
    }

    /**
     * 策略异常
     * <p>相关业务逻辑：平仓失败，可平数量不足
     * <p>mysql锁定顺序： 开关 -> 任务 -> 订单 -> 仓位
     *
     * @param dto dto
     * @since 2022/11/28 10:33
     */
    @Transactional(rollbackFor = Throwable.class)
    public void error(StrategyContractErrorDTO dto) {
        String msgPrefix = "策略异常-";
        LocalDateTime now = LocalDateTime.now();
        Long userId = 0L;
        String username = "系统";

        // 查询并锁定开关
        QueryWrapper<StrategySwitchContract> querySwitch = new QueryWrapper<>();
        querySwitch.lambda()
                .eq(StrategySwitchContract::getId, dto.getSwitchId())
                .in(StrategySwitchContract::getStatus,
                        StrategySwitchContractEnum.Status.PAUSE.getValue(),
                        StrategySwitchContractEnum.Status.RUNNING.getValue())
                .last("for update");
        StrategySwitchContract switchContract = strategySwitchContractMapper.selectOne(querySwitch);
        if (Objects.isNull(switchContract)) {
            throw new MqAckException(msgPrefix + "开关状态异常，switchId：" + dto.getSwitchId());
        }
        // 旧数据
        String oldData = commonUtil.toJson(switchContract);

        switchContract.setRemark(dto.getRemark());
        switchContract.setStatus(StrategySwitchContractEnum.Status.ERROR.getValue());
        switchContract.setUpdateTime(now);
        switchContract.setUpdateBy(userId);
        switchContract.setUpdateName(username);
        if (!switchContract.updateById()) {
            throw new MqNotAckException(msgPrefix + "修改开关状态失败，switchId：" + dto.getSwitchId());
        }

        // 查询运行中的任务，并锁定
        QueryWrapper<StrategyTaskContract> queryTask = new QueryWrapper<>();
        queryTask.lambda()
                .eq(StrategyTaskContract::getSwitchId, dto.getSwitchId())
                .in(StrategyTaskContract::getStatus,
                        StrategyTaskContractEnum.Status.PAUSE.getValue(),
                        StrategyTaskContractEnum.Status.RUNNING.getValue())
                .last("for update");
        StrategyTaskContract task = strategyTaskContractMapper.selectOne(queryTask);
        if (Objects.isNull(task)) {
            throw new MqNotAckException(msgPrefix + "任务状态异常，switchId：" + dto.getSwitchId());
        }

        // 修改任务
        task.setRemark(dto.getRemark());
        if (task.getPendingOrderNum() == 0) {
            task.setErrorSettlementPrice(dto.getSymbolPrice());
            task.setErrorSettlementStatus(StrategyTaskContractEnum.ErrorSettlementStatus.RUNNING.getValue());
        }
        task.setStatus(StrategyTaskContractEnum.Status.ERROR.getValue());
        task.setUpdateTime(now);
        task.setUpdateBy(userId);
        task.setUpdateName(username);
        if (strategyTaskContractMapper.updateById(task) != 1) {
            throw new MqNotAckException(msgPrefix + "修改任务状态失败，taskId：" + task.getId());
        }

        // 没有挂单时，移除任务缓存，结算异常仓位。有挂单时，在同步订单后进行异常结算
        if (task.getPendingOrderNum() == 0) {
            redisUtils.delPersistentCache(RedisKeyEnum.STRATEGY_TASK_CONTRACT, task.getId());
            // 发送mq消息，结算异常仓位
            StrategyContractErrorSettlementDTO errorSettlementDTO = new StrategyContractErrorSettlementDTO();
            errorSettlementDTO.setTaskId(task.getId());
            errorSettlementDTO.setStatus(task.getStatus());
            strategyContractUtil.sendMqErrorSettlement(errorSettlementDTO);
        }

        // 获取策略、策略组缓存
        StrategyContract strategy = strategyContractUtil.getStrategyCache(task.getStrategyId(), false);
        StrategyGroupContract group = strategyContractUtil.getStrategyGroupCache(strategy.getGroupId(), false);
        // 发送mq，策略异常邮件
        strategyContractUtil.sendEmailStrategyError(task.getUserId(), strategy, group, dto.getRemark());
        // 发送mq，系统日志
        SystemLog systemLog = new SystemLog();
        securityUtil.initCreateAbout(systemLog, null, now);
        systemLog.setOperatorType(SystemLogEnum.OperatorType.SYSTEM.getValue());
        systemLog.setType(SystemLogEnum.Type.STRATEGY_ERROR);
        systemLog.setOldData(oldData);
        systemLog.setNewData(commonUtil.toJson(switchContract));
        strategyContractUtil.sendMqSystemLog(systemLog);
    }

    /**
     * 策略异常解决
     * <p>相关业务逻辑：人工确认异常已解决
     * <p>mysql锁定顺序： 开关 -> 任务 -> 订单 -> 仓位
     *
     * @param dto dto
     * @since 2022/11/28 10:33
     */
    @Transactional(rollbackFor = Throwable.class)
    public void errorResolve(StrategyContractErrorResolveDTO dto) {
        String msgPrefix = "策略异常解决-";
        LocalDateTime now = LocalDateTime.now();
        PigxUser user = SecurityUtil.getLoginUserNotNull();

        // 查询异常的任务，并锁定
        QueryWrapper<StrategyTaskContract> queryTask = new QueryWrapper<>();
        queryTask.lambda()
                .eq(StrategyTaskContract::getId, dto.getTaskId())
                .eq(StrategyTaskContract::getStatus, StrategyTaskContractEnum.Status.ERROR.getValue())
                .last("for update");
        StrategyTaskContract task = strategyTaskContractMapper.selectOne(queryTask);
        if (Objects.isNull(task)) {
            throw new MqAckException(msgPrefix + "任务状态异常，taskId：" + dto.getTaskId());
        }
        // 旧数据
        String oldData = commonUtil.toJson(task);

        if (task.getPendingOrderNum() != 0) {
            throw new DuckweedException("该任务下存在委托中的订单，请稍后再试");
        }
        if (Objects.equals(StrategyTaskContractEnum.ErrorSettlementStatus.RUNNING.getValue(), task.getErrorSettlementStatus())) {
            throw new DuckweedException("当前任务正在进行持仓异常部分的利润结算，请稍后再试");
        }

        // 修改任务
        task.setRemark(dto.getRemark());
        task.setStatus(StrategyTaskContractEnum.Status.ERROR_RESOLVED.getValue());
        task.setUpdateTime(now);
        task.setUpdateBy(user.getId());
        task.setUpdateName(user.getName());
        if (strategyTaskContractMapper.updateById(task) != 1) {
            throw new MqAckException(msgPrefix + "修改任务状态失败，taskId：" + dto.getTaskId());
        }

        // 修改开关
        UpdateWrapper<StrategySwitchContract> updateSwitch = new UpdateWrapper<>();
        updateSwitch.lambda()
                .set(StrategySwitchContract::getRemark, dto.getRemark())
                .set(StrategySwitchContract::getStatus, StrategySwitchContractEnum.Status.ERROR_RESOLVED.getValue())
                .set(StrategySwitchContract::getUpdateTime, now)
                .set(StrategySwitchContract::getUpdateBy, user.getId())
                .set(StrategySwitchContract::getUpdateName, user.getName())
                .eq(StrategySwitchContract::getId, task.getSwitchId())
                .eq(StrategySwitchContract::getStatus, StrategySwitchContractEnum.Status.ERROR.getValue());
        if (strategySwitchContractMapper.update(null, updateSwitch) != 1) {
            throw new MqAckException(msgPrefix + "修改开关状态失败，switchId：" + task.getSwitchId());
        }

        // 获取策略、策略组缓存
        StrategyContract strategy = strategyContractUtil.getStrategyCache(task.getStrategyId(), false);
        StrategyGroupContract group = strategyContractUtil.getStrategyGroupCache(strategy.getGroupId(), false);
        // 发送mq，策略异常解决邮件
        strategyContractUtil.sendEmailStrategyErrorResolved(task.getUserId(), strategy, group);
        // 发送mq，系统日志
        SystemLog systemLog = new SystemLog();
        securityUtil.initCreateAbout(systemLog, user, now);
        systemLog.setCreateName(user.getName());
        systemLog.setUpdateName(user.getName());
        systemLog.setOperatorType(SystemLogEnum.OperatorType.MANAGER.getValue());
        systemLog.setType(SystemLogEnum.Type.STRATEGY_ERROR_RESOLVE);
        systemLog.setOldData(oldData);
        systemLog.setNewData(commonUtil.toJson(task));
        strategyContractUtil.sendMqSystemLog(systemLog);
    }

    /**
     * 订单异常
     * <p>相关业务逻辑：同步订单时查询不到订单信息
     * <p>mysql锁定顺序： 开关 -> 任务 -> 订单 -> 仓位
     *
     * @param dto dto
     * @throws MqAckLogException 抛出异常后，syncOrder的catch块会负责回滚事务
     * @since 2022/11/28 10:33
     */
    public void errorOrder(StrategyOrderContractErrorDTO dto) {
        StrategyTaskContractCacheBO taskCache = dto.getTaskCache();
        String msgPrefix = "订单异常-";
        LocalDateTime now = LocalDateTime.now();
        Long userId = 0L;
        String username = "系统";
        StrategyOrderContract order = dto.getOrder();
        boolean isOpen = Objects.equals(StrategyOrderContractEnum.BusinessType.OPEN.getValue(), order.getBusinessType());
        boolean isClose = Objects.equals(StrategyOrderContractEnum.BusinessType.CLOSE.getValue(), order.getBusinessType())
                || Objects.equals(StrategyOrderContractEnum.BusinessType.MANUAL_CLOSE.getValue(), order.getBusinessType())
                || Objects.equals(StrategyOrderContractEnum.BusinessType.MANUAL_CLOSE_PART.getValue(), order.getBusinessType());
        // 开启本地事务
        mysqlUtil.startLocalTx();

        // 修改订单状态
        UpdateWrapper<StrategyOrderContract> updateOrder = new UpdateWrapper<>();
        updateOrder.lambda()
                .set(StrategyOrderContract::getRemark, dto.getRemark())
                .set(StrategyOrderContract::getStatus, StrategyOrderContractEnum.Status.ERROR.getValue())
                .set(StrategyOrderContract::getUpdateTime, now)
                .set(StrategyOrderContract::getUpdateBy, userId)
                .set(StrategyOrderContract::getUpdateName, username)
                .eq(StrategyOrderContract::getId, order.getId())
                .in(StrategyOrderContract::getStatus, StrategyOrderContractEnum.Status.RUNNING.getValue());
        if (strategyOrderContractMapper.update(null, updateOrder) != 1) {
            throw new MqAckLogException(msgPrefix + "修改订单状态失败，orderId：" + order.getId());
        }

        // 查询两个仓位，并锁定
        List<StrategyPositionContract> positionList = strategyContractUtil.getPositionList(order.getTaskId());
        // 根据方向获取订单对应的仓位
        StrategyPositionContract position = StrategyContractUtil.getCurrentPositionBySide(positionList, order.getSide(), false);
        // 根据方向判断仓位状是否正确，止盈止损不需要判断
        if (isOpen && !Objects.equals(StrategyPositionContractEnum.Status.ADDING.getValue(), position.getStatus())
                || isClose && !Objects.equals(StrategyPositionContractEnum.Status.CLOSING.getValue(), position.getStatus())) {
            throw new MqAckLogException("仓位信息异常");
        }

        // 根据当前持仓量更新仓位状态
        String newStatus = position.getNumNow() > 0 ?
                StrategyPositionContractEnum.Status.HOLDING.getValue()
                : StrategyPositionContractEnum.Status.EMPTY.getValue();
        position.setStatus(newStatus);
        position.setUpdateTime(LocalDateTime.now());
        if (!position.updateById()) {
            throw new MqAckLogException("修改仓位记录失败");
        }

        // 查询未完成、未关闭的任务，并锁定
        QueryWrapper<StrategyTaskContract> queryTask = new QueryWrapper<>();
        queryTask.lambda()
                .eq(StrategyTaskContract::getId, order.getTaskId())
                .in(StrategyTaskContract::getStatus,
                        StrategyTaskContractEnum.Status.PAUSE.getValue(),
                        StrategyTaskContractEnum.Status.RUNNING.getValue(),
                        StrategyTaskContractEnum.Status.ERROR.getValue())
                .last("for update");
        StrategyTaskContract task = strategyTaskContractMapper.selectOne(queryTask);
        if (Objects.isNull(task)
                || Objects.isNull(task.getPendingOrderNum())
                || task.getPendingOrderNum() == 0) {
            throw new MqAckLogException(msgPrefix + "任务信息异常，orderId：" + order.getId());
        }
        boolean isTaskError = Objects.equals(StrategyTaskContractEnum.Status.ERROR.getValue(), task.getStatus());

        // 回滚任务记录
        task.setPendingOrderNum(task.getPendingOrderNum() - 1);
        task.setUpdateTime(now);
        task.setUpdateBy(userId);
        task.setUpdateName(username);
        if (isTaskError && task.getPendingOrderNum() == 0) {
            task.setErrorSettlementPrice(order.getPrice());
            task.setErrorSettlementStatus(StrategyTaskContractEnum.ErrorSettlementStatus.RUNNING.getValue());
        }
        if (strategyTaskContractMapper.updateById(task) != 1) {
            throw new MqAckLogException(msgPrefix + "回滚任务信息失败，orderId：" + order.getId());
        }

        // 任务状态为异常，没有挂单删除任务缓存
        if (isTaskError && task.getPendingOrderNum() == 0) {
            // 删除任务缓存
            redisUtils.delPersistentCache(RedisKeyEnum.STRATEGY_TASK_CONTRACT, task.getId());
            // 提交本地事务
            mysqlUtil.commitLocalTx();
            // 删除同步次数累计
            strategyContractUtil.removeSyncCount(order.getId());

            // 发送mq消息，结算异常仓位
            StrategyContractErrorSettlementDTO errorSettlementDTO = new StrategyContractErrorSettlementDTO();
            errorSettlementDTO.setTaskId(task.getId());
            errorSettlementDTO.setStatus(task.getStatus());
            strategyContractUtil.sendMqErrorSettlement(errorSettlementDTO);

        } else {
            // 刷新任务缓存
            StrategyTaskContractCacheBO newTaskCache = strategyContractUtil.flushTaskCacheByEntity(taskCache, task, taskCache.getLastOpenAvg(), positionList);
            // 刷新任务JVM缓存
            dto.setTaskCache(newTaskCache);

            // 提交本地事务
            mysqlUtil.commitLocalTx();
        }

        // 删除订单缓存
        redisUtils.delPersistentCache(RedisKeyEnum.STRATEGY_ORDER_CONTRACT, order.getId());
    }

    /**
     * 止盈止损异常
     * <p>相关业务逻辑：同步止盈止损时查询不到止盈止损信息
     *
     * @param dto        dto
     * @param reason     错误原因
     * @param retryCount 重试次数
     * @throws MqAckLogException 抛出异常后，syncOrder的catch块会负责回滚事务
     * @since 2022/11/28 10:33
     */
    public void errorStopOrder(StrategySyncStopOrderMsgDTO dto, String reason, Long retryCount) {
        try {
            // 没有重试次数，尝试获取redis中的重试次数数据
            if (Objects.isNull(retryCount)) {
                Long retryCountCache = redisUtils.getIncrCache(RedisKeyEnum.STRATEGY_STOP_ORDER_CONTRACT_SYNC_COUNT, dto.getStopOrderId(), Long.class);

                if (Objects.isNull(retryCountCache)) {
                    retryCount = 0L;
                } else {
                    retryCount = Long.valueOf(retryCountCache.toString());
                }
            }

            // 开启本地事务
            mysqlUtil.startLocalTx();

            // 修改任务数据
            UpdateWrapper<StrategyTaskContract> updateTask = new UpdateWrapper<>();
            updateTask.lambda()
                    .setSql("sync_stop_order_num = sync_stop_order_num - 1")
                    .eq(StrategyTaskContract::getId, dto.getTaskId())
                    .ge(StrategyTaskContract::getSyncStopOrderNum, 1)
                    .in(StrategyTaskContract::getStatus,
                            StrategyTaskContractEnum.Status.PAUSE.getValue(),
                            StrategyTaskContractEnum.Status.RUNNING.getValue());
            if (strategyTaskContractMapper.update(null, updateTask) != 1) {
                throw new MqAckLogException("修改止盈止损为异常时，修改任务失败");
            }

            // 修改止盈止损数据
            UpdateWrapper<StrategyStopOrderContract> updateStopOrder = new UpdateWrapper<>();
            updateStopOrder.lambda()
                    .set(StrategyStopOrderContract::getSyncCount, retryCount)
                    .set(StrategyStopOrderContract::getSyncStatus, StrategyStopOrderContractEnum.SyncStatus.DONE.getValue())
                    .set(StrategyStopOrderContract::getStatus, StrategyStopOrderContractEnum.Status.ERROR.getValue())
                    .set(StrategyStopOrderContract::getRemark, reason)
                    .eq(StrategyStopOrderContract::getId, dto.getStopOrderId())
                    .eq(StrategyStopOrderContract::getSyncStatus, StrategyStopOrderContractEnum.SyncStatus.RUNNING.getValue())
                    .eq(StrategyStopOrderContract::getStatus, StrategyStopOrderContractEnum.Status.NOT_ACTIVE.getValue());
            if (strategyStopOrderContractMapper.update(null, updateStopOrder) != 1) {
                throw new MqAckLogException("修改止盈止损为异常时，修改止盈止损失败");
            }

            // 获取任务缓存
            StrategyTaskContractCacheBO taskCache = strategyContractUtil.getStrategyTaskCache(dto.getTaskId(), false);
            StrategyStopOrderContractCacheBO stopOrderCache = taskCache.getStopOrderCache(dto.getStopOrderId());

            // 修改仓位数据
            UpdateWrapper<StrategyPositionContract> updatePosition = new UpdateWrapper<>();
            updatePosition.lambda()
                    .setSql("num_stop_order = num_stop_order - " + stopOrderCache.getNum())
                    .ge(StrategyPositionContract::getNumStopOrder, stopOrderCache.getNum())
                    .eq(StrategyPositionContract::getId, stopOrderCache.getId());
            if (strategyPositionContractMapper.update(null, updatePosition) != 1) {
                throw new MqAckLogException("修改止盈止损为异常时，修改仓位失败");
            }
            // 修改仓位缓存
            StrategyPositionContract positionCache = StrategyContractUtil.getCurrentPositionBySide(taskCache.getPositionList(), stopOrderCache.getSide(), false);
            positionCache.setNumStopOrder(positionCache.getNumStopOrder() - stopOrderCache.getNum());

            // 减少同步中止盈止损订单数量
            taskCache.setSyncStopOrderNum(taskCache.getSyncStopOrderNum() - 1);
            if (taskCache.getSyncStopOrderNum() < 0) {
                throw new MqNotAckException("修改止盈止损为异常时，任务缓存SyncStopOrderNum数据异常：" + taskCache.getSyncStopOrderNum());
            }

            // 将异常的止盈止损从缓存中去除
            taskCache.removeStopOrderCache(dto.getStopOrderId());
            // 刷新任务缓存
            redisUtils.flushPersistentCache(RedisKeyEnum.STRATEGY_TASK_CONTRACT, taskCache.getId(), taskCache);
            // 删除计数缓存
            if (retryCount > 0) {
                redisUtils.delIncrCache(RedisKeyEnum.STRATEGY_STOP_ORDER_CONTRACT_SYNC_COUNT, dto.getStopOrderId());
            }

            // 提交本地事务
            mysqlUtil.commitLocalTx();
        } catch (Throwable e) {
            // 回滚本地事务
            mysqlUtil.rollbackLocalTx();
            throw e;
        }
    }

    /**
     * 检查燃油币是否达到警戒值
     *
     * @param userId 用户id
     * @since 2022/12/11 21:10
     */
    @RabbitListener(queues = StrategyMqConstant.QUEUE_FANOUT_STRATEGY_CONTRACT_FUEL_WARNING)
    public void checkFuelWarning(Long userId) {
        try {
            // 查询用户燃油币额度
            BigDecimal fuelBalance = strategyContractUtil.getFuelBalance(userId);

            // 开启本地事务
            mysqlUtil.startLocalTx();

            // 查询运行中的开关，并锁定
            QueryWrapper<StrategySwitchContract> querySwitch = new QueryWrapper<>();
            querySwitch.lambda()
                    .eq(StrategySwitchContract::getUserId, userId)
                    .eq(StrategySwitchContract::getStatus, StrategySwitchContractEnum.Status.RUNNING.getValue())
                    .last("for update");
            StrategySwitchContract switchContract = strategySwitchContractMapper.selectOne(querySwitch);
            if (Objects.isNull(switchContract)) {
                // 回滚本地事务
                mysqlUtil.rollbackLocalTx();
                return;
            }

            // 策略组预扣燃油费为0时，不再进行判断
            StrategyContract strategy = strategyContractUtil.getStrategyCache(switchContract.getStrategyId(), false);
            StrategyGroupContract group = strategyContractUtil.getStrategyGroupCache(strategy.getGroupId(), false);
            if (group.getFuelFrozen().compareTo(BigDecimal.ZERO) == 0) {
                // 回滚本地事务
                mysqlUtil.rollbackLocalTx();
                return;
            }

            // 判断余额是否低于等于警戒值
            BigDecimal warningRatio = strategyContractUtil.getFuelWarningRatio();
            BigDecimal warningValue = group.getFuelFrozen().multiply(BigDecimalUtils.toPercent(warningRatio));
            boolean needWarning = warningValue.compareTo(fuelBalance) >= 0;
            // 低于预警值 && 没有发送过邮件
            if (needWarning && Objects.equals(CommonDictEnum.YesOrNo.NO.getValue(), switchContract.getSendFuelReminder())) {
                // 高于预警值 && 发送过邮件
                switchContract.setSendFuelReminder(CommonDictEnum.YesOrNo.YES.getValue());
                if (!switchContract.updateById()) {
                    throw new MqNotAckException("修改开关燃油币告警邮件发送状态失败");
                }

                // 提交本地事务
                mysqlUtil.commitLocalTx();

                // 发送余额告警邮件
                strategyContractUtil.sendEmailStrategyFuelWarning(switchContract.getUserId(),
                        strategy,
                        group,
                        fuelBalance,
                        warningRatio,
                        group.getFuelFrozen());

            } else if (!needWarning && Objects.equals(CommonDictEnum.YesOrNo.YES.getValue(), switchContract.getSendFuelReminder())) {
                // 高于预警值 && 发送过邮件
                switchContract.setSendFuelReminder(CommonDictEnum.YesOrNo.NO.getValue());
                if (!switchContract.updateById()) {
                    throw new MqNotAckException("修改开关燃油币告警邮件发送状态失败");
                }

                // 提交本地事务
                mysqlUtil.commitLocalTx();

            } else {
                // 不需要任何处理，回滚本地事务
                mysqlUtil.rollbackLocalTx();
            }
        } catch (Throwable e) {
            // 回滚本地事务
            mysqlUtil.rollbackLocalTx();
            throw e;
        }
    }

    /**
     * 任务异常/用户手动关闭之后结算持仓异常部分的利润
     *
     * @param dto dto
     * @since 2022/12/11 21:10
     */
    @RabbitListener(queues = StrategyMqConstant.QUEUE_FANOUT_STRATEGY_TASK_CONTRACT_ERROR_SETTLEMENT)
    @Transactional(rollbackFor = Throwable.class)
    public void errorSettlement(StrategyContractErrorSettlementDTO dto) {
        // 查询任务并锁定
        QueryWrapper<StrategyTaskContract> queryTask = new QueryWrapper<>();
        queryTask.lambda()
                .eq(StrategyTaskContract::getId, dto.getTaskId())
                .eq(StrategyTaskContract::getStatus, dto.getStatus())
                .eq(StrategyTaskContract::getPendingOrderNum, 0)
                .eq(StrategyTaskContract::getErrorSettlementStatus, StrategyTaskContractEnum.ErrorSettlementStatus.RUNNING.getValue())
                .last("for update");
        StrategyTaskContract task = strategyTaskContractMapper.selectOne(queryTask);
        if (Objects.isNull(task)) {
            throw new MqNotAckException("未能查询到异常任务信息");
        }

        try {
            errorSettlement(dto, task);

        } catch (Throwable e) {
            // 回滚全局事务
            SeataUtils.rollbackGlobalTx();
            e.printStackTrace();
            throw e;
        }
    }

    /**
     * 任务异常/用户手动关闭之后结算持仓异常部分的利润
     *
     * @param dto  dto
     * @param task 任务
     * @since 2022/12/11 21:10
     */
    public void errorSettlement(StrategyContractErrorSettlementDTO dto, StrategyTaskContract task) {
        // 获取策略信息
        StrategyContract strategy = strategyContractUtil.getStrategyCache(task.getStrategyId(), false);
        // 获取产品信息
        SystemProduct product = strategyContractUtil.getSystemProductCache(strategy.getProductId(), false);

        // 查询仓位并锁定，此时仓位应只会是无持仓、持仓中、已平仓三种状态之一
        QueryWrapper<StrategyPositionContract> queryPosition = new QueryWrapper<>();
        queryPosition.lambda()
                .eq(StrategyPositionContract::getTaskId, dto.getTaskId())
                .in(StrategyPositionContract::getStatus,
                        StrategyPositionContractEnum.Status.EMPTY.getValue(),
                        StrategyPositionContractEnum.Status.HOLDING.getValue(),
                        StrategyPositionContractEnum.Status.CLOSE.getValue())
                .last("for update");
        List<StrategyPositionContract> positionList = strategyPositionContractMapper.selectList(queryPosition);
        if (CollectionUtils.isEmpty(positionList)) {
            throw new MqNotAckException("未能查询到异常任务信息的仓位信息");
        }
        // 判断是否拥有持仓
        boolean hasPosition = positionList.stream()
                .anyMatch(i -> Objects.equals(StrategyPositionContractEnum.Status.HOLDING.getValue(), i.getStatus()));
        if (!hasPosition) {
            // 修改任务为已结算
            task.setErrorSettlementStatus(StrategyTaskContractEnum.ErrorSettlementStatus.COMPLETE.getValue());
            if (!task.updateById()) {
                throw new MqNotAckException("修改任务状态失败");
            }
            // 直接结束
            return;
        }

        // 计算异常部分仓位的利润
        BigDecimal errorSettlementProfit = BigDecimal.ZERO;
        for (StrategyPositionContract position : positionList) {
            // 不是持仓中状态
            if (!Objects.equals(StrategyPositionContractEnum.Status.HOLDING.getValue(), position.getStatus())) {
                continue;
            }

            // 根据当前持仓计算利润
            BigDecimal profit = StrategyOrderContractCalculator.profit(position.getSide(),
                    task.getErrorSettlementPrice(),
                    position.getAvgOpen(),
                    position.getNumNow(),
                    product.getSheetValue());
            position.setErrorSettlementProfit(profit);
            position.setStatus(StrategyPositionContractEnum.Status.ERROR_SETTLEMENT.getValue());
            if (!position.updateById()) {
                throw new MqNotAckException("修改仓位状态失败");
            }

            errorSettlementProfit = errorSettlementProfit.add(profit);
        }

        // 计算临时总净利润
        BigDecimal tempNetProfit = task.getProfit().add(errorSettlementProfit)
                .add(task.getHandlingFee());
        // 如果有利润，计算燃油费消耗
        TransactionThenDTO profitSettlementMsgDto = null;
        if (tempNetProfit.compareTo(BigDecimal.ZERO) > 0) {
            // 获取策略组信息
            StrategyGroupContract group = strategyContractUtil.getStrategyGroupCache(strategy.getGroupId(), false);
            BigDecimal platformPart = StrategyTaskContractCalculator.profitSettlement(tempNetProfit, group.getPlatformRatio());
            task.setFuelCost(task.getFuelCost().add(platformPart));

            // 开启全局事务
            SeataUtils.startGlobalTx();
            // 扣除利润结算
            BigDecimal newFuelBalance = strategyContractUtil.updateFuel(task.getUserId(), FuelConsumeEnum.Type.PROFIT_SETTLEMENT, platformPart);

            // 生成制度消息
            profitSettlementMsgDto = new TransactionThenDTO();
            profitSettlementMsgDto.setUserId(task.getUserId());
            profitSettlementMsgDto.setStrategyName(StrategyContractUtil.getStrategyName(group, strategy));
            profitSettlementMsgDto.setFuelType(FuelConsumeEnum.Type.PROFIT_SETTLEMENT);
            profitSettlementMsgDto.setFuel(platformPart);
            profitSettlementMsgDto.setFuelBalance(newFuelBalance);
            profitSettlementMsgDto.setGrossProfit(task.getProfit().add(errorSettlementProfit));
            profitSettlementMsgDto.setNetProfit(tempNetProfit);
        }

        // 修改任务信息
        task.setErrorSettlementProfit(errorSettlementProfit);
        task.setErrorSettlementStatus(StrategyTaskContractEnum.ErrorSettlementStatus.COMPLETE.getValue());
        if (!task.updateById()) {
            throw new MqNotAckException("修改任务状态失败");
        }

        // 修改开关信息
        UpdateWrapper<StrategySwitchContract> updateSwitch = new UpdateWrapper<>();
        updateSwitch.lambda()
                .set(StrategySwitchContract::getErrorSettlementProfit, errorSettlementProfit)
                .eq(StrategySwitchContract::getId, task.getSwitchId())
                .eq(StrategySwitchContract::getStatus, dto.getStatus());
        if (strategySwitchContractMapper.update(null, updateSwitch) != 1) {
            throw new MqNotAckException("修改开关状态失败");
        }

        // 提交全局事务
        SeataUtils.commitGlobalTx();

        // 如果制度消息不为null，发送消息
        if (!Objects.isNull(profitSettlementMsgDto)) {
            strategyContractUtil.sendMqDividend(profitSettlementMsgDto);
        }
    }

    /**
     * 止盈止损触发时，构建下单参数
     *
     * @param taskId            任务id
     * @param side              方向（0做多、1做空）
     * @param actualTriggerSide 实际触发方向（0止盈、1止损
     * @param actualNum         实际委托数量
     * @param actualPrice       实际委托价格
     * @param positionId        仓位id
     * @return {@link PlaceOrderBO}
     * @since 2023/2/9 18:42
     */
    public PlaceOrderBO buildWhenStopOrderTrigger(Long taskId,
                                                  String side,
                                                  String actualTriggerSide,
                                                  Integer actualNum,
                                                  BigDecimal actualPrice,
                                                  Long positionId) {
        TaskRunner taskRunner = new TaskRunner(redisUtils, strategyOrderContractHandlerManager, taskId, false, false);
        return taskRunner.buildWhenStopOrderTrigger(side,
                actualTriggerSide,
                actualNum,
                actualPrice,
                positionId);
    }

    /**
     * 撤销任务的所有止盈止损
     *
     * @param dto dto
     * @since 2023/2/10 14:59
     */
    public void cancelByTask(StrategyStopOrderContractCancelBatchDTO dto) {
        StrategyTaskContractCacheBO task = strategyContractUtil.getStrategyTaskCache(dto.getTaskId(), false);
        if (CollectionUtils.isEmpty(task.getStopOrderList())) {
            throw new DuckweedException("该任务下没有找到可以撤销的止盈止损");
        }

        List<Long> stopOrderIdList = task.getStopOrderList().stream()
                .map(StrategyStopOrderContractCacheBO::getId)
                .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(stopOrderIdList)) {
            throw new DuckweedException("该任务下没有找到可以撤销的止盈止损");
        }

        dto.setIdList(stopOrderIdList);
        strategyStopOrderContractHandlerManager.cancelBatch(dto);
    }

    /**
     * mq消息创建止盈止损
     *
     * @param dto dto
     * @since 2022/11/28 10:32
     */
    @RabbitListener(queues = StrategyMqConstant.QUEUE_FANOUT_STRATEGY_STOP_ORDER_CONTRACT_CREATE)
    public void stopOrderAddMq(StrategyStopOrderContractAddDTO dto) {
        try {
            strategyStopOrderContractHandlerManager.place(dto);
        } catch (MqAckLogException e) {
            log.error(e.getMessage());
        } catch (MqAckException ignore) {

        }
    }

    /**
     * mq消息撤销止盈止损
     *
     * @param dto dto
     * @since 2022/11/28 10:32
     */
    @RabbitListener(queues = StrategyMqConstant.QUEUE_FANOUT_STRATEGY_STOP_ORDER_CONTRACT_CANCEL)
    public void stopOrderCancelMq(StrategyStopOrderContractCancelBatchDTO dto) {
        try {
            cancelByTask(dto);
        } catch (MqAckLogException e) {
            log.error(e.getMessage());
        } catch (MqAckException ignore) {

        }
    }

    /**
     * 任务处理器
     *
     * @since 2022/11/28 17:53
     */
    private class TaskRunner {

        private final RedisUtils redisUtils;
        private final StrategyOrderContractHandlerManager strategyOrderContractHandlerManager;

        private final StrategyTaskContractCacheBO task;
        private final StrategyContract strategy;
        private final BigDecimal price;

        private StrategyGroupContract group;
        private SystemPlatform platform;
        private SystemProduct product;

        /**
         * 利润信息
         */
        private StrategyTaskContractCalculateProfitBO profitInfo;
        /**
         * 仓位补仓张数
         */
        private int positionAddNum;
        /**
         * 是否检查能否下单
         */
        boolean checkCanPlaceOrder;
        /**
         * 是否检查止盈止损
         */
        boolean checkStopOrder;

        /**
         * 构造函数
         *
         * @since 2022/12/2 18:43
         */
        public TaskRunner(RedisUtils redisUtils,
                          StrategyOrderContractHandlerManager strategyOrderContractHandlerManager,
                          Long id,
                          boolean checkCanPlaceOrder,
                          boolean checkStopOrder) {
            this.redisUtils = redisUtils;
            this.strategyOrderContractHandlerManager = strategyOrderContractHandlerManager;
            this.checkCanPlaceOrder = checkCanPlaceOrder;
            this.checkStopOrder = checkStopOrder;

            // 获取任务
            task = strategyContractUtil.getStrategyTaskCache(id, false);
            // 获取策略
            strategy = strategyContractUtil.getStrategyCache(task.getStrategyId(), false);
            // 获取产品
            getProduct();
            // 获取价格
            price = getPrice(strategy.getSymbolName(), getProduct().getPlatformId());
            // 判断是否可下单
            if (checkCanPlaceOrder) {
                runnable();
            }

            // System.err.println("上次成交均价：" + task.getLastOpenAvg());
            if (Objects.isNull(price)) {
                throw new MqAckLogException(String.format("币对 %s 价格数据为空", strategy.getSymbolName()));
            }
        }

        /**
         * 获取币对最新价
         *
         * @param symbolName 币对名称
         * @return {@link BigDecimal}
         * @since 2022/11/28 19:31
         */
        private BigDecimal getPrice(String symbolName, Long platform) {
            // 查询价格
            BigDecimal lastPrice = null;
            if (platform.equals(SystemPlatformConstant.OKX)) {
                lastPrice = redisUtils.getSymbolPrice(strategy.getSymbolName(), CommonRedisConstants.OKX + CommonRedisConstants.LAST_PRICE);
            }
            if (Objects.isNull(lastPrice)) {
                String errorMsg = String.format("币对价格为空：%s", symbolName);
                throw new MqAckLogException(errorMsg);
            }

            return lastPrice;

        }

        /**
         * 判断是否可执行
         *
         * @since 2022/11/28 19:13
         */
        private void runnable() {
            // 平仓导致异常时，不再运行
            if (Objects.equals(StrategyTaskContractEnum.Status.ERROR.getValue(), task.getStatus())) {
                throw new MqAckException("任务处于异常状态中");
            }

            // todo 修改旧缓存，下次更新去除
            if (Objects.isNull(task.getSyncStopOrderNum())) {
                task.setSyncStopOrderNum(0);
            }

            // 有挂单、同步中止盈止损时不处理
            if (task.getPendingOrderNum() != 0 || task.getSyncStopOrderNum() != 0) {
                if (checkStopOrder) {
                    // 检查止盈止损
                    checkExternalStopProfitLost();
                }

                throw new MqAckException("任务下存在未处理完成的订单/止盈止损");
            }
        }

        /**
         * 开仓
         *
         * @return {@link boolean} true 执行了业务逻辑
         * @since 2022/11/28 19:49
         */
        public boolean openPosition() {
            // 下单数量为0时，是首次开仓
            if (task.getOrderCount() != 0) {
                return false;
            }

            // 组装开仓参数
            List<PlaceOrderBO> paramList = buildOpenParam();
            // 下单
            strategyOrderContractHandlerManager.placeOrder(paramList);

            return true;
        }

        /**
         * 检查设置的第三方止盈止损是否触发
         *
         * @return {@link boolean}
         * @since 2023/2/8 19:56
         */
        public boolean checkExternalStopProfitLost() {
            List<StrategyStopOrderContractCacheBO> stopOrderList = task.getStopOrderList();
            if (CollectionUtils.isEmpty(stopOrderList)) {
                return false;
            }

            // 记录缓存数据是否发生变化
            AtomicBoolean hasChange = new AtomicBoolean(false);
            // 遍历获取新被触发的止盈止损id
            List<StrategyStopOrderContractCacheBO> triggerList = stopOrderList.stream()
                    .filter(stopOrder -> {
                        if (!Objects.equals(StrategyStopOrderContractEnum.SyncStatus.NOT_SYNC.getValue(), stopOrder.getSyncStatus())) {
                            return false;
                        }

                        // 刷新止盈止损价格记录
                        if (flushStopOrderPriceRecord(stopOrder)) {
                            hasChange.set(true);
                        }

                        // 止盈
                        if (Objects.equals(CommonDictEnum.YesOrNo.YES.getValue(), stopOrder.getHasStopProfit())) {
                            // 做多
                            if (Objects.equals(StrategyOrderContractEnum.Side.LONG.getValue(), stopOrder.getSide())
                                    && stopOrder.getMaxPrice().compareTo(stopOrder.getStopProfitTrigger()) >= 0) {
                                stopOrder.setSyncStatus(StrategyStopOrderContractEnum.SyncStatus.RUNNING.getValue());
                                hasChange.set(true);
                                return true;

                                // 做空
                            } else if (Objects.equals(StrategyOrderContractEnum.Side.SHORT.getValue(), stopOrder.getSide())
                                    && stopOrder.getMinPrice().compareTo(stopOrder.getStopProfitTrigger()) <= 0) {
                                stopOrder.setSyncStatus(StrategyStopOrderContractEnum.SyncStatus.RUNNING.getValue());
                                hasChange.set(true);
                                return true;
                            }
                        }

                        // 止损
                        if (Objects.equals(CommonDictEnum.YesOrNo.YES.getValue(), stopOrder.getHasStopLost())) {
                            // 做多
                            if (Objects.equals(StrategyOrderContractEnum.Side.LONG.getValue(), stopOrder.getSide())
                                    && stopOrder.getMinPrice().compareTo(stopOrder.getStopLostTrigger()) <= 0) {
                                stopOrder.setSyncStatus(StrategyStopOrderContractEnum.SyncStatus.RUNNING.getValue());
                                hasChange.set(true);
                                return true;
                                // 做空
                            } else if (Objects.equals(StrategyOrderContractEnum.Side.SHORT.getValue(), stopOrder.getSide())
                                    && stopOrder.getMaxPrice().compareTo(stopOrder.getStopLostTrigger()) >= 0) {
                                stopOrder.setSyncStatus(StrategyStopOrderContractEnum.SyncStatus.RUNNING.getValue());
                                hasChange.set(true);
                                return true;
                            }
                        }

                        return false;
                    }).collect(Collectors.toList());

            // 没有需要触发的，直接结束
            if (triggerList.isEmpty()) {
                // 如果缓存发生了变化，更新缓存
                if (hasChange.get()) {
                    flushCache();
                }

                return false;
            }

            // 修改止盈止损同步状态，发送mq消息，同步止盈止损信息
            handleStopOrderTrigger(triggerList);
            return true;
        }

        /**
         * 更新止盈止损价格记录
         *
         * @param cache 缓存
         * @return {@link boolean}
         * @since 2023/2/9 10:08
         */
        private boolean flushStopOrderPriceRecord(StrategyStopOrderContractCacheBO cache) {
            boolean hasChange = false;
            if (Objects.isNull(cache.getMaxPrice())
                    || cache.getMaxPrice().compareTo(price) < 0) {
                cache.setMaxPrice(price);
                hasChange = true;
            }
            if (Objects.isNull(cache.getMinPrice())
                    || cache.getMinPrice().compareTo(price) > 0) {
                cache.setMinPrice(price);
                hasChange = true;
            }

            return hasChange;
        }

        /**
         * 处理止盈止损触发
         *
         * @param triggerList 触发的止盈止损
         * @since 2023/2/9 09:28
         */
        private void handleStopOrderTrigger(List<StrategyStopOrderContractCacheBO> triggerList) {
            try {
                List<Long> idList = triggerList.stream()
                        .map(StrategyStopOrderContractCacheBO::getId)
                        .collect(Collectors.toList());

                // 开启本地事务
                mysqlUtil.startLocalTx();

                // 修改任务数据
                UpdateWrapper<StrategyTaskContract> updateTask = new UpdateWrapper<>();
                updateTask.lambda()
                        .setSql("sync_stop_order_num = sync_stop_order_num + " + triggerList.size())
                        .eq(StrategyTaskContract::getId, task.getId());
                if (strategyTaskContractMapper.update(null, updateTask) != 1) {
                    throw new MqAckLogException("触发止盈止损时，修改任务状态失败");
                }
                // 修改止盈止损
                UpdateWrapper<StrategyStopOrderContract> updateStopOrder = new UpdateWrapper<>();
                updateStopOrder.lambda()
                        .set(StrategyStopOrderContract::getSyncStatus, StrategyStopOrderContractEnum.SyncStatus.RUNNING.getValue())
                        .eq(StrategyStopOrderContract::getSyncStatus, StrategyStopOrderContractEnum.SyncStatus.NOT_SYNC.getValue())
                        .eq(StrategyStopOrderContract::getStatus, StrategyStopOrderContractEnum.Status.NOT_ACTIVE.getValue())
                        .in(StrategyStopOrderContract::getId, idList);
                if (strategyStopOrderContractMapper.update(null, updateStopOrder) != triggerList.size()) {
                    throw new MqAckLogException("修改触发的止盈止损状态失败");
                }

                // 修改redis缓存
                flushCache();
                // 提交本地事务
                mysqlUtil.commitLocalTx();

                // 发送mq消息，同步止盈止损
                triggerList.forEach(stopOrder -> {
                    StrategySyncStopOrderMsgDTO dto = new StrategySyncStopOrderMsgDTO();
                    dto.setPlatformId(stopOrder.getPlatformId());
                    dto.setExternalId(stopOrder.getExternalId());
                    dto.setStopOrderId(stopOrder.getId());
                    dto.setStrategyId(strategy.getId());
                    dto.setSwitchId(task.getSwitchId());
                    dto.setTaskId(task.getId());
                    dto.setUserId(stopOrder.getUserId());
                    dto.setSideType(stopOrder.getSideType());

                    strategyContractUtil.sendMqSyncStopOrder(dto);
                });

            } catch (Throwable e) {
                // 回滚本地事务
                mysqlUtil.rollbackLocalTx();
                e.printStackTrace();
            }
        }

        /**
         * 止盈止损
         *
         * @return {@link boolean} true 执行了业务逻辑
         * @since 2022/11/28 20:01
         */
        public boolean stopProfitLost() {
            // 计算收益
            profitInfo = strategyOrderContractHandlerManager.calculateProfit(task,
                    strategy,
                    product,
                    price);
            // 根据盈亏情况判断是否需要平仓
            boolean needClose = checkProfit();

            // 需要平仓
            if (needClose) {
                // 组装平仓参数
                List<PlaceOrderBO> paramList = buildCloseParam();
                // 下单
                strategyOrderContractHandlerManager.placeOrder(paramList);

            } else {
                // 不需要，更新最高盈利率、最高价格、最低价格数据
                task.setHighestPrice(price.compareTo(task.getHighestPrice()) > 0 ?
                        price : task.getHighestPrice());
                task.setLowestPrice(task.getLowestPrice().compareTo(BigDecimal.ZERO) == 0 || price.compareTo(task.getLowestPrice()) < 0 ?
                        price : task.getLowestPrice());
                task.setHighestProfit(profitInfo.getProfit().compareTo(task.getHighestProfit()) > 0 ?
                        profitInfo.getProfit() : task.getHighestProfit());
                task.setHighestProfitRate(profitInfo.getProfitRate().compareTo(task.getHighestProfitRate()) > 0 ?
                        profitInfo.getProfitRate() : task.getHighestProfitRate());
            }

            return needClose;
        }

        /**
         * 更新缓存
         *
         * @since 2022/12/16 14:56
         */
        private void flushCache() {
            // 更新缓存
            redisUtils.flushPersistentCache(RedisKeyEnum.STRATEGY_TASK_CONTRACT, task.getId(), task);
        }

        /**
         * 根据盈亏情况判断是否需要平仓
         *
         * @return {@link boolean}
         * @since 2022/11/29 14:47
         */
        private boolean checkProfit() {
            // 盈利 >= 止盈数值（可能为空）
            if (profitInfo.getIsProfit()
                    && !Objects.isNull(strategy.getStopProfitValue())
                    && profitInfo.getProfit().compareTo(strategy.getStopProfitValue()) >= 0) {
                return true;

            } else if (!profitInfo.getIsProfit()
                    && profitInfo.getProfit().negate().compareTo(strategy.getStopLostValue()) >= 0) {
                // 亏损 >= 止损数值
                return true;

            } else if (task.getHighestProfitRate().compareTo(strategy.getStopProfitRatio()) > 0
                    && profitInfo.getProfitRate().compareTo(strategy.getStopProfitRatio()) >= 0) {
                // 最高盈利率 > 止盈比例 && 当前盈利率 >= 止盈比例，计算回降比例
                BigDecimal fallbackRatio = StrategyTaskContractCalculator.fallbackRatio(profitInfo.getProfitRate(),
                        task.getHighestProfitRate(),
                        strategy.getStopProfitRatio());
                if (fallbackRatio.compareTo(strategy.getCallbackRatio()) >= 0) {
                    return true;
                }

                // 最高盈利率 >= 盈利率 && 止盈比例 >= 盈利率
                // if (task.getHighestProfitRate().compareTo(profitInfo.getProfitRate()) >= 0
                //         && strategy.getStopProfitRatio().compareTo(profitInfo.getProfitRate()) >= 0) {
                //     return true;
                // }
            }

            return false;
        }

        /**
         * 判断补仓
         *
         * @return {@link boolean} true 执行了业务逻辑
         * @since 2022/11/28 20:01
         */
        public boolean addPosition() {
            // 判断任务是否暂停
            if (Objects.equals(StrategyTaskContractEnum.Status.PAUSE.getValue(), task.getStatus())) {
                return false;
            }

            // 判断补仓次数
            if (task.getAddCount() >= strategy.getOrderNum()) {
                return false;
            }

            // 价格 < 价格波动范围最小值（可能为空） || 价格 > 价格波动范围最大值（可能为空），超出范围不补仓
            if (!Objects.isNull(strategy.getMinPrice()) && price.compareTo(strategy.getMinPrice()) < 0
                    || !Objects.isNull(strategy.getMaxPrice()) && price.compareTo(strategy.getMaxPrice()) > 0) {
                return false;
            }

            // 当前价格波动比
            BigDecimal currentFluctuationRatio = StrategyTaskContractCalculator.currentFluctuationRatio(price, task.getLastOpenAvg());
            // 价格波动比例（%）
            BigDecimal fluctuationRatio = StrategyTaskContractCalculator.fluctuationRatio(strategy.getFluctuationRatio(),
                    strategy.getFluctuationIncrease(),
                    task.getAddCount());
            int highestCompareZero = task.getHighestFluctuationRatio().compareTo(BigDecimal.ZERO);
            // 最高波动比例为0时，更新记录
            if (highestCompareZero == 0) {
                task.setHighestFluctuationRatio(currentFluctuationRatio);
                return false;
            }

            // 回调比例
            BigDecimal callbackRatio = StrategyTaskContractCalculator.callbackRatio(currentFluctuationRatio, task.getHighestFluctuationRatio());

            // 补仓方向，为null表示不需要补仓
            String addSide = null;
            // 首次补仓
            if (task.getAddCount() == 0) {
                // 最高波动记录绝对值小于价格波动比例时，更新绝对值更大的值为最高波动记录
                BigDecimal currentFluctuationRatioAbs = currentFluctuationRatio.abs();
                BigDecimal highestFluctuationRatioAbs = task.getHighestFluctuationRatio().abs();
                if (highestFluctuationRatioAbs.compareTo(strategy.getFluctuationRatio()) < 0) {
                    if (currentFluctuationRatioAbs.compareTo(highestFluctuationRatioAbs) >= 0) {
                        task.setHighestFluctuationRatio(currentFluctuationRatio);
                        return false;
                    }

                } else if (highestCompareZero > 0) {
                    // 上涨时，当前波动比例数值更大或相等，更新记录
                    if (currentFluctuationRatio.compareTo(task.getHighestFluctuationRatio()) >= 0) {
                        task.setHighestFluctuationRatio(currentFluctuationRatio);
                        return false;

                    } else if (callbackRatio.compareTo(strategy.getCallbackRatio()) >= 0) {
                        // 回调比例大于等于配置，上涨补空
                        addSide = StrategyOrderContractEnum.Side.SHORT.getValue();
                    }

                } else {
                    // 下跌时，当前波动比例数值更小或相等，更新记录
                    if (currentFluctuationRatio.compareTo(task.getHighestFluctuationRatio()) <= 0) {
                        task.setHighestFluctuationRatio(currentFluctuationRatio);
                        return false;

                    } else if (callbackRatio.compareTo(strategy.getCallbackRatio()) >= 0) {
                        // 回调比例大于等于配置,下跌补多
                        addSide = StrategyOrderContractEnum.Side.LONG.getValue();
                    }
                }

            } else {
                // 不是首次补仓，根据首次补仓方向记录亏损侧最大波动比例
                boolean isLong = StrategyContractUtil.isLong(task.getFirstAddSide());
                // 多仓判断价格下跌幅度
                if (isLong) {
                    // 当前跌幅更高，更新记录
                    if (currentFluctuationRatio.compareTo(task.getHighestFluctuationRatio()) <= 0) {
                        task.setHighestFluctuationRatio(currentFluctuationRatio);
                        return false;

                    } else if (task.getHighestFluctuationRatio().negate().compareTo(fluctuationRatio) >= 0
                            && callbackRatio.compareTo(strategy.getCallbackRatio()) >= 0) {
                        // 最高跌幅符合价格波动比例，回调比例大于等于配置
                        addSide = StrategyOrderContractEnum.Side.LONG.getValue();
                    }

                } else {
                    // 空仓判断价格上涨幅度
                    // 当前涨幅更高，更新记录
                    if (currentFluctuationRatio.compareTo(task.getHighestFluctuationRatio()) >= 0) {
                        task.setHighestFluctuationRatio(currentFluctuationRatio);
                        return false;

                    } else if (task.getHighestFluctuationRatio().compareTo(fluctuationRatio) >= 0
                            && callbackRatio.compareTo(strategy.getCallbackRatio()) >= 0) {
                        // 最高涨幅符合价格波动比例，回调比例大于等于配置
                        addSide = StrategyOrderContractEnum.Side.SHORT.getValue();
                    }
                }
            }

            // 不需要补仓，直接结束
            if (Strings.isBlank(addSide)) {
                return false;
            }

            // 下单参数集合
            List<PlaceOrderBO> paramList = new ArrayList<>(2);
            // 构建补仓下单参数
            PlaceOrderBO addParam = buildAddParam(addSide);
            paramList.add(addParam);

            // 补仓后判断是否需要锁仓
            int currentAddCount = task.getAddCount() + 1;
            // System.err.println("第" + currentAddCount + "次补仓，补仓数量：" + addParam.getNum());
            if (!Objects.isNull(strategy.getLockStart())
                    && currentAddCount >= strategy.getLockStart()
                    && !Objects.isNull(strategy.getLockRatio())
                    && strategy.getLockRatio().compareTo(BigDecimal.ZERO) > 0) {

                // 获取补仓相对的方向
                String lockSide = Objects.equals(StrategyOrderContractEnum.Side.LONG.getValue(), addSide) ?
                        StrategyOrderContractEnum.Side.SHORT.getValue()
                        : StrategyOrderContractEnum.Side.LONG.getValue();

                // 构建锁仓下单参数
                PlaceOrderBO lockParam = buildLockParam(lockSide);
                paramList.add(lockParam);
                // System.err.println("第" + currentAddCount + "次补仓，锁仓数量：" + lockParam.getNum());
            }

            // 下单
            strategyOrderContractHandlerManager.placeOrder(paramList);

            return true;
        }

        /**
         * 获得策略组数据
         *
         * @return {@link StrategyGroupContract}
         * @since 2022/12/2 17:17
         */
        private StrategyGroupContract getGroup() {
            if (!Objects.isNull(group)) {
                return group;
            }

            Long id = strategy.getGroupId();
            group = strategyContractUtil.getStrategyGroupCache(id, false);

            return group;
        }

        /**
         * 获得平台数据
         *
         * @return {@link SystemPlatform}
         * @since 2022/12/2 17:17
         */
        private SystemPlatform getPlatform() {
            if (!Objects.isNull(platform)) {
                return platform;
            }

            Long id = strategy.getPlatformId();
            platform = strategyContractUtil.getSystemPlatformCache(id, false);

            return platform;
        }

        /**
         * 获得产品数据
         *
         * @return {@link SystemProduct}
         * @since 2022/12/2 17:17
         */
        private SystemProduct getProduct() {
            if (!Objects.isNull(product)) {
                return product;
            }

            Long id = strategy.getProductId();
            product = strategyContractUtil.getSystemProductCache(id, false);

            return product;
        }

        /**
         * 构建开仓下单参数
         *
         * @return {@link List< PlaceOrderBO>}
         * @since 2022/12/2 17:18
         */
        private List<PlaceOrderBO> buildOpenParam() {
            // System.err.println("下单-开仓");

            // 下单数量
            int orderNum = 1;
            // 首仓方向 = 双向开仓，同时开多空两个仓，先下多单
            if (Objects.equals(StrategyContractEnum.Side.DOUBLE_SIDE.getValue(), strategy.getSide())) {
                orderNum = 2;

            } else if (!Objects.equals(StrategyContractEnum.Side.LONG.getValue(), strategy.getSide())
                    && !Objects.equals(StrategyContractEnum.Side.SHORT.getValue(), strategy.getSide())) {
                String errorMsg = String.format("无法识别的首仓方向 %s，策略id：%s", strategy.getSide(), strategy.getId());
                throw new MqAckLogException(errorMsg);
            }

            List<PlaceOrderBO> list = new ArrayList<>(orderNum);
            for (int i = 0; i < orderNum; i++) {
                PlaceOrderBO param = buildCommonParam();
                param.setBusinessType(StrategyOrderContractEnum.BusinessType.OPEN.getValue());
                param.setFirstOrder(list.size() == 0);

                // 首仓方向 = 单向，直接使用策略的方向，字典值一致
                if (orderNum == 1) {
                    param.setSide(strategy.getSide());
                } else {
                    // 首仓方向 = 双向，先下多单，后下空单
                    param.setSide(i == 0 ?
                            StrategyOrderContractEnum.Side.LONG.getValue()
                            : StrategyOrderContractEnum.Side.SHORT.getValue());
                }

                param.setOperationType(StrategyOrderContractEnum.OperationType.OPEN.getValue());
                param.setNum(strategy.getFirstNum());
                param.setNoDealNum(param.getNum());

                list.add(param);
            }

            return list;

        }

        /**
         * 构建通用下单参数
         *
         * @return {@link PlaceOrderBO}
         * @since 2022/12/2 17:50
         */
        public PlaceOrderBO buildCommonParam() {
            LocalDateTime now = LocalDateTime.now();
            PlaceOrderBO param = new PlaceOrderBO();

            param.setId(commonUtil.getId());
            param.setRetryCount(0);
            param.setTaskId(task.getId());
            param.setUserId(task.getUserId());
            param.setPlatformId(strategy.getPlatformId());
            param.setProductName(getProduct().getName());
            param.setPrice(price);
            param.setType(StrategyOrderContractEnum.Type.MARKET.getValue());
            param.setPositionMode(StrategyOrderContractEnum.PositionMode.CROSS.getValue());
            param.setOrderValue(BigDecimal.ZERO);
            param.setSheetValue(product.getSheetValue());
            param.setMultiple(strategy.getMultiple());
            param.setDealNum(0);
            param.setDealAvg(BigDecimal.ZERO);
            param.setHandlingFee(BigDecimal.ZERO);
            param.setMargin(BigDecimal.ZERO);
            param.setProfit(BigDecimal.ZERO);
            param.setFuelHandlingFee(BigDecimal.ZERO);
            param.setCreateTime(now);
            param.setCreateBy(0L);
            param.setCreateName("系统");
            param.setUpdateTime(now);
            param.setUpdateBy(0L);
            param.setUpdateName("系统");
            param.setStatus(StrategyOrderContractEnum.Status.RUNNING.getValue());
            param.setIsDeleted(CommonDictEnum.IsDeleted.NO.getValue());

            param.setTaskCache(task);

            return param;
        }

        /**
         * 构建补仓下单参数
         *
         * @return {@link PlaceOrderBO}
         * @since 2022/12/2 17:18
         */
        private PlaceOrderBO buildAddParam(String side) {
            // System.err.println("下单-补仓");
            // 计算当前补仓次数
            int currentAddCount = task.getAddCount() + 1;
            positionAddNum = strategy.getFirstNum() << currentAddCount;

            // 补仓肯定是第一个参数
            PlaceOrderBO param = buildCommonParam();
            param.setBusinessType(StrategyOrderContractEnum.BusinessType.ADD.getValue());
            param.setSide(side);
            param.setOperationType(StrategyOrderContractEnum.OperationType.OPEN.getValue());
            param.setNum(positionAddNum);
            param.setNoDealNum(param.getNum());

            return param;
        }

        /**
         * 构建锁仓下单参数
         *
         * @return {@link PlaceOrderBO}
         * @since 2022/12/2 17:18
         */
        private PlaceOrderBO buildLockParam(String side) {
            // System.err.println("下单-锁仓");
            PlaceOrderBO param = buildCommonParam();
            param.setBusinessType(StrategyOrderContractEnum.BusinessType.LOCK.getValue());
            // 锁仓肯定是第二个参数，在补仓之后
            param.setFirstOrder(false);

            param.setSide(side);
            param.setOperationType(StrategyOrderContractEnum.OperationType.OPEN.getValue());

            // 补仓数量的一定百分比
            param.setNum(StrategyTaskContractCalculator.lockNum(positionAddNum, strategy.getLockRatio()));
            param.setNoDealNum(param.getNum());

            return param;
        }

        /**
         * 构建平仓下单参数
         *
         * @return {@link List< PlaceOrderBO>}
         * @since 2022/12/2 17:18
         */
        private List<PlaceOrderBO> buildCloseParam() {
            // System.err.println("下单-平仓");

            // 同时最多有两个运行中的仓位
            List<PlaceOrderBO> list = new ArrayList<>(2);
            task.getPositionList().forEach(i -> {
                if (i.getNumNow() == 0) {
                    return;
                }

                PlaceOrderBO param = buildCommonParam();
                // 是第一个订单
                param.setFirstOrder(list.size() == 0);
                param.setOperationType(StrategyOrderContractEnum.OperationType.CLOSE.getValue());
                param.setBusinessType(StrategyOrderContractEnum.BusinessType.CLOSE.getValue());
                param.setSide(i.getSide());
                param.setNum(i.getNumNow());
                param.setNoDealNum(param.getNum());

                list.add(param);
            });

            return list;
        }

        /**
         * 构建手动开仓参数
         *
         * @param side 方向
         * @param num  数量
         * @return {@link PlaceOrderBO}
         * @since 2023/1/17 14:12
         */
        public PlaceOrderBO buildOpenManualParam(String side, Integer num) {
            PlaceOrderBO param = buildCommonParam();
            param.setBusinessType(StrategyOrderContractEnum.BusinessType.MANUAL_OPEN.getValue());
            param.setSide(side);
            param.setOperationType(StrategyOrderContractEnum.OperationType.OPEN.getValue());
            param.setNum(num);
            param.setNoDealNum(param.getNum());

            return param;
        }

        /**
         * 构建手动平仓下单参数
         *
         * @return {@link List< PlaceOrderBO>}
         * @since 2022/12/2 17:18
         */
        private List<PlaceOrderBO> buildCloseManualParam() {
            // System.err.println("下单-平仓");

            // 同时最多有两个运行中的仓位
            List<PlaceOrderBO> list = new ArrayList<>(2);
            task.getPositionList().forEach(i -> {
                if (i.getNumNow() == 0) {
                    return;
                }

                PlaceOrderBO param = buildCommonParam();
                // 是第一个订单
                param.setFirstOrder(list.size() == 0);
                param.setOperationType(StrategyOrderContractEnum.OperationType.CLOSE.getValue());
                param.setBusinessType(StrategyOrderContractEnum.BusinessType.MANUAL_CLOSE.getValue());
                param.setSide(i.getSide());
                param.setNum(i.getNumNow());
                param.setNoDealNum(param.getNum());

                list.add(param);
            });

            return list;
        }

        /**
         * 构建手动部分平仓下单参数
         *
         * @param side 方向
         * @param num  数量
         * @return {@link PlaceOrderBO}
         * @since 2022/12/2 17:18
         */
        private PlaceOrderBO buildClosePartParam(String side, Integer num) {
            // System.err.println("下单-手动部分平仓");

            StrategyPositionContract position = StrategyContractUtil.getCurrentPositionBySide(task.getPositionList(), side, true);
            if (Objects.isNull(position)) {
                throw new MqAckLogException("仓位信息不存在");
            } else if (!Objects.equals(StrategyPositionContractEnum.Status.HOLDING.getValue(), position.getStatus())) {
                throw new MqAckLogException("仓位状态已改变");
            } else if (position.getNumNow() - position.getNumStopOrder() < num) {
                throw new MqAckLogException("仓位可平数量不足，当前可平数量：" + (position.getNumNow() - position.getNumStopOrder()));
            }

            PlaceOrderBO param = buildCommonParam();
            // 一定是第一个订单
            param.setOperationType(StrategyOrderContractEnum.OperationType.CLOSE.getValue());
            param.setBusinessType(StrategyOrderContractEnum.BusinessType.MANUAL_CLOSE_PART.getValue());
            param.setSide(position.getSide());
            param.setNum(num);
            param.setNoDealNum(param.getNum());

            return param;
        }

        /**
         * 止盈止损触发时，构建下单参数，用来持久化到mysql2
         *
         * @param side              方向（0做多、1做空）
         * @param actualTriggerSide 实际触发方向（0止盈、1止损
         * @param actualNum         实际委托数量
         * @param actualPrice       实际委托价格
         * @param positionId        仓位id
         * @return {@link PlaceOrderBO}
         * @since 2023/2/9 18:25
         */
        private PlaceOrderBO buildWhenStopOrderTrigger(String side,
                                                       String actualTriggerSide,
                                                       Integer actualNum,
                                                       BigDecimal actualPrice,
                                                       Long positionId) {
            PlaceOrderBO param = buildCommonParam();
            param.setBusinessType(StrategyStopOrderContractEnum.ActualTriggerSide.PROFIT.getValue().equals(actualTriggerSide) ?
                    StrategyOrderContractEnum.BusinessType.STOP_PROFIT.getValue()
                    : StrategyOrderContractEnum.BusinessType.STOP_LOST.getValue());
            param.setSide(side);
            param.setOperationType(StrategyOrderContractEnum.OperationType.CLOSE.getValue());
            param.setNum(actualNum);
            param.setNoDealNum(param.getNum());
            // 覆盖当前价格
            param.setPrice(actualPrice);
            param.setPositionId(positionId);
            return param;
        }
    }
}
